test_runner.py (3519B)
1 import asyncio 2 import os 3 import re 4 import shutil 5 import tarfile 6 import tempfile 7 import unittest 8 9 import responses 10 11 from condprof import client 12 from condprof.client import ROOT_URL, TC_SERVICE 13 from condprof.main import main 14 15 client.RETRIES = 1 16 client.RETRY_PAUSE = 0 17 GECKODRIVER = os.path.join(os.path.dirname(__file__), "fakegeckodriver.py") 18 FIREFOX = os.path.join(os.path.dirname(__file__), "fakefirefox.py") 19 CHANGELOG = re.compile(ROOT_URL + "/.*/changelog.json") 20 FTP = "https://ftp.mozilla.org/pub/firefox/nightly/latest-mozilla-central/" 21 PROFILE = re.compile(ROOT_URL + "/.*/.*tgz") 22 23 24 with open(os.path.join(os.path.dirname(__file__), "ftp_mozilla.html")) as f: 25 FTP_PAGE = f.read() 26 27 FTP_ARCHIVE = re.compile( 28 "https://ftp.mozilla.org/pub/firefox/nightly/latest-mozilla-central/firefox.*" 29 ) 30 31 ADDON = re.compile("https://addons.mozilla.org/.*/.*xpi") 32 33 34 async def fakesleep(duration): 35 duration = min(duration, 1) 36 await asyncio.realsleep(duration) 37 38 39 asyncio.realsleep = asyncio.sleep 40 asyncio.sleep = fakesleep 41 42 43 class TestRunner(unittest.TestCase): 44 def setUp(self): 45 self.archive_dir = tempfile.mkdtemp() 46 responses.add(responses.GET, CHANGELOG, json={"error": "not found"}, status=404) 47 responses.add( 48 responses.GET, FTP, content_type="text/html", body=FTP_PAGE, status=200 49 ) 50 51 profile_tgz = os.path.join(os.path.dirname(__file__), "profile.tgz") 52 profile = os.path.join(os.path.dirname(__file__), "profile") 53 54 with tarfile.open(profile_tgz, "w:gz") as tar: 55 tar.add(profile, arcname=".") 56 57 with open(profile_tgz, "rb") as f: 58 PROFILE_DATA = f.read() 59 60 os.remove(profile_tgz) 61 62 responses.add( 63 responses.GET, 64 FTP_ARCHIVE, 65 body="1", 66 headers={"content-length": "1"}, 67 status=200, 68 ) 69 70 responses.add( 71 responses.GET, 72 PROFILE, 73 body=PROFILE_DATA, 74 headers={"content-length": str(len(PROFILE_DATA))}, 75 status=200, 76 ) 77 78 responses.add( 79 responses.HEAD, 80 PROFILE, 81 body="", 82 headers={"content-length": str(len(PROFILE_DATA))}, 83 status=200, 84 ) 85 86 responses.add(responses.HEAD, FTP_ARCHIVE, body="", status=200) 87 88 responses.add( 89 responses.GET, ADDON, body="1", headers={"content-length": "1"}, status=200 90 ) 91 92 responses.add( 93 responses.HEAD, ADDON, body="", headers={"content-length": "1"}, status=200 94 ) 95 96 responses.add(responses.HEAD, TC_SERVICE, body="", status=200) 97 98 def tearDown(self): 99 shutil.rmtree(self.archive_dir) 100 101 @responses.activate 102 def test_runner(self): 103 if "FXA_USERNAME" not in os.environ: 104 os.environ["FXA_USERNAME"] = "me" 105 if "FXA_PASSWORD" not in os.environ: 106 os.environ["FXA_PASSWORD"] = "password" 107 try: 108 args = [ 109 "--geckodriver", 110 GECKODRIVER, 111 "--firefox", 112 FIREFOX, 113 self.archive_dir, 114 ] 115 main(args) 116 # XXX we want a bunch of assertions here to check 117 # that the archives dir gets filled correctly 118 finally: 119 if os.environ["FXA_USERNAME"] == "me": 120 del os.environ["FXA_USERNAME"] 121 if os.environ["FXA_PASSWORD"] == "password": 122 del os.environ["FXA_PASSWORD"]