test_client.py (4121B)
1 import json 2 import os 3 import re 4 import shutil 5 import tarfile 6 import tempfile 7 import unittest 8 9 import responses 10 from mozprofile.prefs import Preferences 11 12 from condprof.client import ROOT_URL, TC_SERVICE, get_profile 13 from condprof.util import _DEFAULT_SERVER 14 15 PROFILE = re.compile(ROOT_URL + "/.*/.*tgz") 16 PROFILE_FOR_TESTS = os.path.join(os.path.dirname(__file__), "profile") 17 SECRETS = re.compile(_DEFAULT_SERVER + "/.*") 18 SECRETS_PROXY = re.compile("http://taskcluster/secrets/.*") 19 20 21 class TestClient(unittest.TestCase): 22 def setUp(self): 23 self.profile_dir = tempfile.mkdtemp() 24 25 # creating profile.tgz on the fly for serving it 26 profile_tgz = os.path.join(self.profile_dir, "profile.tgz") 27 with tarfile.open(profile_tgz, "w:gz") as tar: 28 tar.add(PROFILE_FOR_TESTS, arcname=".") 29 30 # self.profile_data is the tarball we're sending back via HTTP 31 with open(profile_tgz, "rb") as f: 32 self.profile_data = f.read() 33 34 self.target = tempfile.mkdtemp() 35 self.download_dir = os.path.expanduser("~/.condprof-cache") 36 if os.path.exists(self.download_dir): 37 shutil.rmtree(self.download_dir) 38 39 responses.add( 40 responses.GET, 41 PROFILE, 42 body=self.profile_data, 43 headers={"content-length": str(len(self.profile_data)), "ETag": "'12345'"}, 44 status=200, 45 ) 46 47 responses.add( 48 responses.HEAD, 49 PROFILE, 50 body="", 51 headers={"content-length": str(len(self.profile_data)), "ETag": "'12345'"}, 52 status=200, 53 ) 54 55 responses.add(responses.HEAD, TC_SERVICE, body="", status=200) 56 57 secret = {"secret": {"username": "user", "password": "pass"}} 58 secret = json.dumps(secret) 59 for pattern in (SECRETS, SECRETS_PROXY): 60 responses.add( 61 responses.GET, 62 pattern, 63 body=secret, 64 headers={"content-length": str(len(secret))}, 65 status=200, 66 ) 67 68 def tearDown(self): 69 shutil.rmtree(self.target) 70 shutil.rmtree(self.download_dir) 71 shutil.rmtree(self.profile_dir) 72 73 @responses.activate 74 def test_cache(self): 75 download_dir = os.path.expanduser("~/.condprof-cache") 76 if os.path.exists(download_dir): 77 num_elmts = len(os.listdir(download_dir)) 78 else: 79 num_elmts = 0 80 81 get_profile(self.target, "win64", "settled", "default") 82 83 # grabbing a profile should generate two files 84 self.assertEqual(len(os.listdir(download_dir)), num_elmts + 2) 85 86 # we do at least two network calls when getting a file, 87 # a HEAD and a GET and possibly a TC secret 88 self.assertTrue(len(responses.calls) >= 2) 89 90 # reseting the response counters 91 responses.calls.reset() 92 93 # and we should reuse them without downloading the file again 94 get_profile(self.target, "win64", "settled", "default") 95 96 # grabbing a profile should not download new stuff 97 self.assertEqual(len(os.listdir(download_dir)), num_elmts + 2) 98 99 # and do a single extra HEAD call, everything else is cached, 100 # even the TC secret 101 self.assertEqual(len(responses.calls), 2) 102 103 prefs_js = os.path.join(self.target, "prefs.js") 104 prefs = Preferences.read_prefs(prefs_js) 105 106 # check that the gfx.blacklist prefs where cleaned out 107 for name, value in prefs: 108 self.assertFalse(name.startswith("gfx.blacklist")) 109 110 # check that we have the startupScanScopes option forced 111 prefs = dict(prefs) 112 self.assertEqual(prefs["extensions.startupScanScopes"], 1) 113 114 # make sure we don't have any marionette option set 115 user_js = os.path.join(self.target, "user.js") 116 for name, value in Preferences.read_prefs(user_js): 117 self.assertFalse(name.startswith("marionette.")) 118 119 120 if __name__ == "__main__": 121 try: 122 import mozunit 123 except ImportError: 124 pass 125 else: 126 mozunit.main(runwith="unittest")