cp3_test.py

#!/usr/bin/env python
import socket
import requests
import os
import time
from grader import grader, tester
import hashlib
import random
from subprocess import Popen, PIPE, STDOUT
import os.path
import signal

def test_daemonization(self):
    if self.testsuite.scores['server_start'] != 1:
        self.skipTest("server failed to start. skip this test")
    
    try:
        f = open('%slisod.lock'%self.testsuite.tmp_dir, 'r')
    except IOError:
        raise Exception("Lockfile does not exist!")
    
    pid = f.readline().strip()
    try:
        pid = int(pid)
    except ValueError:
        raise Exception("Lockfile does not have a valid pid!")

    if pid <= 0:
        raise Exception("Lockfile does have an invalid pid!")
        print "Server running on pid %d" % pid

    try:
        os.kill(pid, 0)
    except OSError:
        Exception("But pid %d is dead or never lived before!" % pid)

    self.testsuite.server_pid = pid
    self.testsuite.scores['test_daemonization'] = 1
    return

def test_pipelining_keepalive(self):
    if self.testsuite.scores['test_daemonization'] != 1:
        self.skipTest("server failed to start. skip this test")
        print "Testing pipelining"

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect(('127.0.0.1', self.testsuite.port))
    s.settimeout(4)
    pipe = "HEAD /index.html HTTP/1.1\r\nHost: 127.0.0.1:%d\r\nConnection: Keep-Alive\r\n\r\n"\%self.testsuite.port*5
    s.send(pipe)
    
    while True:
        try:
            buf = s.recv(1024)
        except socket.timeout:
            self.testsuite.scores['test_pipelining_keepalive'] += 0.5
            break
        
        if buf == "":
            print "Server connection does not keepalive!"
            break
    s.close()

    print "Testing pipelining with Connection: Close"
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect(('127.0.0.1', self.testsuite.port))
    s.settimeout(4)
    pipe2 = pipe+"HEAD /index.html HTTP/1.1\r\nHost: 127.0.0.1:%d\r\nConnection: Close\r\n\r\n"%self.testsuite.port
    s.send(pipe2)
    
    while True:
        try:
            buf = s.recv(1024)
        except socket.timeout:
            print "Server doesn't close connections as requested!"
            break
        
        if buf == "":
            self.testsuite.scores['test_pipelining_keepalive'] += 0.5
            break
    s.close()
    if self.testsuite.scores['test_pipelining_keepalive'] < 1:
    raise Exception("Failed in some of the testcases of keepalive")

def test_invalidPUT(self):
    print '----- Testing PUT via SSL -----'
    if self.testsuite.scores['test_daemonization'] != 1:
        self.skipTest("server failed to start. skip this test")

    time.sleep(1)
    for test in self.testsuite.SSL_tests:
        try:
            response = requests.put(test % self.testsuite.tls_port,\
            timeout=3.0, verify=SIGNER_CERT)
        except requests.exceptions.SSLError:
            raise Exception("Failed to connect via SSL!")

    self.pAssertEqual(501, response.status_code)
    self.testsuite.scores['test_invalidPUT'] = 1

def test_invalidLENGTH(self):
    print '----- Testing Bad Length Post via http-----'
    if self.testsuite.scores['test_daemonization'] != 1:
        self.skipTest("server failed to start. skip this test")

    s = requests.Session()
    prepped = requests.Request('POST', 'http://127.0.0.1:%d/cgi/' \
                                % self.testsuite.port, data=BAD_POST_DATA, \
                                headers={'Connection':'Close'}).prepare()
    prepped.headers['Content-Length'] = -1000
    response = s.send(prepped, timeout=10.0)
    print response.status_code
    reasonable_codes = [400, 411, 413]
    self.pAssertEqual(True, response.status_code in reasonable_codes)
    self.testsuite.scores['test_invalidLENGTH'] = 1