<s> """<STR_LIT>""" <EOL> import BaseHTTPServer <EOL> import logging <EOL> import os <EOL> import sys <EOL> import threading <EOL> import time <EOL> import unittest <EOL> import uuid <EOL> import gflags <EOL> FLAGS = gflags . FLAGS <EOL> from dpxdt import server <EOL> from dpxdt . client import capture_worker <EOL> from dpxdt . client import workers <EOL> from dpxdt . server import db <EOL> from dpxdt . server import models <EOL> from dpxdt . tools import run_server <EOL> from dpxdt . tools import site_diff <EOL> import test_utils <EOL> server_thread = None <EOL> def setUpModule ( ) : <EOL> """<STR_LIT>""" <EOL> global server_thread <EOL> server_thread = test_utils . start_server ( ) <EOL> run_server . run_workers ( ) <EOL> def create_build ( ) : <EOL> """<STR_LIT>""" <EOL> build = models . Build ( name = '<STR_LIT>' ) <EOL> db . session . add ( build ) <EOL> db . session . commit ( ) <EOL> return build . id <EOL> def wait_for_release ( build_id , release_name , timeout_seconds = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> start = time . time ( ) <EOL> while True : <EOL> release = ( models . Release . query <EOL> . filter_by ( build_id = build_id , name = release_name ) <EOL> . order_by ( models . Release . number . desc ( ) ) <EOL> . first ( ) ) <EOL> db . session . refresh ( release ) <EOL> if release . status == models . Release . REVIEWING : <EOL> return release <EOL> else : <EOL> logging . info ( '<STR_LIT>' , release . status ) <EOL> assert time . time ( ) - start < timeout_seconds , ( <EOL> '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> def webserver ( func ) : <EOL> """<STR_LIT>""" <EOL> class HandlerClass ( BaseHTTPServer . BaseHTTPRequestHandler ) : <EOL> def do_GET ( self ) : <EOL> output = func ( self . path ) <EOL> if output : <EOL> code , content_type , result = output <EOL> else : <EOL> code , content_type , result = <NUM_LIT> , '<STR_LIT>' , '<STR_LIT>' <EOL> self . send_response ( code ) <EOL> self . send_header ( '<STR_LIT:Content-Type>' , content_type ) <EOL> self . end_headers ( ) <EOL> if result : <EOL> self . wfile . write ( result ) <EOL> server = BaseHTTPServer . HTTPServer ( ( '<STR_LIT>' , <NUM_LIT:0> ) , HandlerClass ) <EOL> thread = threading . Thread ( target = server . serve_forever ) <EOL> thread . daemon = True <EOL> thread . start ( ) <EOL> server . server_prefix = '<STR_LIT>' % server . server_address [ <NUM_LIT:1> ] <EOL> logging . info ( '<STR_LIT>' , server . server_prefix ) <EOL> return server <EOL> class SiteDiffTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . coordinator = workers . get_coordinator ( ) <EOL> self . build_id = create_build ( ) <EOL> self . release_name = uuid . uuid4 ( ) . hex <EOL> self . client = server . app . test_client ( ) <EOL> def testEndToEnd ( self ) : <EOL> """<STR_LIT>""" <EOL> @ webserver <EOL> def test1 ( path ) : <EOL> if path == '<STR_LIT:/>' : <EOL> return <NUM_LIT:200> , '<STR_LIT>' , '<STR_LIT>' <EOL> logging . info ( '<STR_LIT>' ) <EOL> site_diff . real_main ( <EOL> start_url = test1 . server_prefix + '<STR_LIT:/>' , <EOL> upload_build_id = self . build_id , <EOL> upload_release_name = self . release_name ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> release = wait_for_release ( self . build_id , self . release_name ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> resp = self . client . get ( <EOL> '<STR_LIT>' % ( <EOL> self . build_id , release . name , release . number ) , <EOL> follow_redirects = True ) <EOL> self . assertEquals ( '<STR_LIT>' , resp . status ) <EOL> self . assertIn ( '<STR_LIT>' , resp . data ) <EOL> self . assertIn ( '<STR_LIT>' , resp . data ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> resp = self . client . post ( <EOL> '<STR_LIT>' , <EOL> data = dict ( <EOL> id = self . build_id , <EOL> name = release . name , <EOL> number = release . number , <EOL> good = True ) , <EOL> follow_redirects = True ) <EOL> self . assertEquals ( '<STR_LIT>' , resp . status ) <EOL> @ webserver <EOL> def test2 ( path ) : <EOL> if path == '<STR_LIT:/>' : <EOL> return <NUM_LIT:200> , '<STR_LIT>' , '<STR_LIT>' <EOL> logging . info ( '<STR_LIT>' ) <EOL> site_diff . real_main ( <EOL> start_url = test2 . server_prefix + '<STR_LIT:/>' , <EOL> upload_build_id = self . build_id , <EOL> upload_release_name = self . release_name ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> release = wait_for_release ( self . build_id , self . release_name ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> resp = self . client . get ( <EOL> '<STR_LIT>' % ( <EOL> self . build_id , release . name , release . number ) , <EOL> follow_redirects = True ) <EOL> self . assertEquals ( '<STR_LIT>' , resp . status ) <EOL> self . assertIn ( '<STR_LIT>' , resp . data ) <EOL> self . assertIn ( '<STR_LIT>' , resp . data ) <EOL> self . assertIn ( '<STR_LIT>' , resp . data ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> site_diff . real_main ( <EOL> start_url = test1 . server_prefix + '<STR_LIT:/>' , <EOL> upload_build_id = self . build_id , <EOL> upload_release_name = self . release_name ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> release = wait_for_release ( self . build_id , self . release_name ) <EOL> test1 . shutdown ( ) <EOL> test2 . shutdown ( ) <EOL> logging . info ( '<STR_LIT>' ) <EOL> resp = self . client . get ( <EOL> '<STR_LIT>' % ( <EOL> self . build_id , release . name , release . number ) , <EOL> follow_redirects = True ) <EOL> self . assertEquals ( '<STR_LIT>' , resp . status ) <EOL> self . assertIn ( '<STR_LIT>' , resp . data ) <EOL> self . assertIn ( '<STR_LIT>' , resp . data ) <EOL> def testCrawler ( self ) : <EOL> """<STR_LIT>""" <EOL> @ webserver <EOL> def test ( path ) : <EOL> if path == '<STR_LIT:/>' : <EOL> return <NUM_LIT:200> , '<STR_LIT>' , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> elif path == '<STR_LIT>' : <EOL> return <NUM_LIT:200> , '<STR_LIT>' , '<STR_LIT>' <EOL> elif path == '<STR_LIT>' : <EOL> return <NUM_LIT> , '<STR_LIT>' , '<STR_LIT>' <EOL> elif path == '<STR_LIT>' : <EOL> return <NUM_LIT:200> , '<STR_LIT>' , '<STR_LIT>' <EOL> site_diff . real_main ( <EOL> start_url = test . server_prefix + '<STR_LIT:/>' , <EOL> upload_build_id = self . build_id , <EOL> upload_release_name = self . release_name , <EOL> ignore_prefixes = [ '<STR_LIT>' ] ) <EOL> release = wait_for_release ( self . build_id , self . release_name ) <EOL> run_list = models . Run . query . all ( ) <EOL> found = set ( run . name for run in run_list ) <EOL> expected = set ( [ '<STR_LIT:/>' , '<STR_LIT>' ] ) <EOL> self . assertEquals ( expected , found ) <EOL> test . shutdown ( ) <EOL> class HtmlRewritingTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testAll ( self ) : <EOL> """<STR_LIT>""" <EOL> base_url = '<STR_LIT>' <EOL> def test ( test_url ) : <EOL> data = '<STR_LIT>' % test_url <EOL> result = site_diff . extract_urls ( base_url , data ) <EOL> if not result : <EOL> return None <EOL> return list ( result ) [ <NUM_LIT:0> ] <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT:/>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( None , test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> self . assertIsNone ( test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( '<STR_LIT>' , <EOL> test ( '<STR_LIT>' ) ) <EOL> scriptTag = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEquals ( <EOL> set ( [ <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ] ) , <EOL> site_diff . extract_urls ( base_url , scriptTag ) ) <EOL> spacesInTag = "<STR_LIT>" <EOL> self . assertEquals ( <EOL> set ( [ '<STR_LIT>' ] ) , <EOL> site_diff . extract_urls ( base_url , spacesInTag ) ) <EOL> jsText = "<STR_LIT>" <EOL> self . assertEquals ( <EOL> set ( [ <EOL> '<STR_LIT>' <EOL> ] ) , <EOL> site_diff . extract_urls ( base_url , jsText ) ) <EOL> def main ( argv ) : <EOL> gflags . MarkFlagAsRequired ( '<STR_LIT>' ) <EOL> gflags . MarkFlagAsRequired ( '<STR_LIT>' ) <EOL> try : <EOL> argv = FLAGS ( argv ) <EOL> except gflags . FlagsError , e : <EOL> print '<STR_LIT>' % ( e , sys . argv [ <NUM_LIT:0> ] , FLAGS ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> test_utils . debug_log_everything ( ) <EOL> unittest . main ( argv = argv ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( sys . argv ) </s>
