cts_utils_test.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586
  1. #!/usr/bin/env vpython3
  2. # Copyright 2019 The Chromium Authors. All rights reserved.
  3. # Use of this source code is governed by a BSD-style license that can be
  4. # found in the LICENSE file.
  5. import os
  6. import re
  7. import tempfile
  8. import shutil
  9. import sys
  10. import unittest
  11. import zipfile
  12. import six
  13. from mock import patch # pylint: disable=import-error
  14. sys.path.append(
  15. os.path.join(
  16. os.path.dirname(__file__), os.pardir, os.pardir, 'third_party',
  17. 'catapult', 'common', 'py_utils'))
  18. # pylint: disable=wrong-import-position,import-error
  19. from py_utils import tempfile_ext
  20. import cts_utils
  21. CIPD_DATA = {}
  22. CIPD_DATA['template'] = """# Copyright notice.
  23. # cipd create instructions.
  24. package: %s
  25. description: Dummy Archive
  26. data:
  27. - file: %s
  28. - file: %s
  29. - file: %s
  30. - file: %s
  31. """
  32. CIPD_DATA['package'] = 'chromium/android_webview/tools/cts_archive'
  33. CIPD_DATA['file1'] = 'arch1/platform1/file1.zip'
  34. CIPD_DATA['file1_arch'] = 'arch1'
  35. CIPD_DATA['file1_platform'] = 'platform1'
  36. CIPD_DATA['file2'] = 'arch1/platform2/file2.zip'
  37. CIPD_DATA['file3'] = 'arch2/platform1/file3.zip'
  38. CIPD_DATA['file4'] = 'arch2/platform2/file4.zip'
  39. CIPD_DATA['yaml'] = CIPD_DATA['template'] % (
  40. CIPD_DATA['package'], CIPD_DATA['file1'], CIPD_DATA['file2'],
  41. CIPD_DATA['file3'], CIPD_DATA['file4'])
  42. CONFIG_DATA = {}
  43. CONFIG_DATA['json'] = """{
  44. "platform1": {
  45. "git": {
  46. "tag_prefix": "platform-1.0"
  47. },
  48. "arch": {
  49. "arch1": {
  50. "filename": "arch1/platform1/file1.zip",
  51. "_origin": "https://a1.p1/f1.zip",
  52. "unzip_dir": "arch1/path/platform1_r1"
  53. },
  54. "arch2": {
  55. "filename": "arch2/platform1/file3.zip",
  56. "_origin": "https://a2.p1/f3.zip",
  57. "unzip_dir": "arch1/path/platform1_r1"
  58. }
  59. },
  60. "test_runs": [
  61. {
  62. "apk": "p1/test.apk"
  63. }
  64. ]
  65. },
  66. "platform2": {
  67. "git": {
  68. "tag_prefix": "platform-2.0"
  69. },
  70. "arch": {
  71. "arch1": {
  72. "filename": "arch1/platform2/file2.zip",
  73. "_origin": "https://a1.p2/f2.zip",
  74. "unzip_dir": "arch1/path/platform2_r1"
  75. },
  76. "arch2": {
  77. "filename": "arch2/platform2/file4.zip",
  78. "_origin": "https://a2.p2/f4.zip",
  79. "unzip_dir": "arch1/path/platform2_r1"
  80. }
  81. },
  82. "test_runs": [
  83. {
  84. "apk": "p2/test1.apk",
  85. "additional_apks": [
  86. {
  87. "apk": "p2/additional_apk_a_1.apk"
  88. }
  89. ]
  90. },
  91. {
  92. "apk": "p2/test2.apk",
  93. "additional_apks": [
  94. {
  95. "apk": "p2/additional_apk_b_1.apk",
  96. "forced_queryable": true
  97. },
  98. {
  99. "apk": "p2/additional_apk_b_2.apk"
  100. }
  101. ]
  102. }
  103. ]
  104. }
  105. }
  106. """
  107. CONFIG_DATA['origin11'] = 'https://a1.p1/f1.zip'
  108. CONFIG_DATA['base11'] = 'f1.zip'
  109. CONFIG_DATA['file11'] = 'arch1/platform1/file1.zip'
  110. CONFIG_DATA['origin12'] = 'https://a2.p1/f3.zip'
  111. CONFIG_DATA['base12'] = 'f3.zip'
  112. CONFIG_DATA['file12'] = 'arch2/platform1/file3.zip'
  113. CONFIG_DATA['apk1'] = 'p1/test.apk'
  114. CONFIG_DATA['origin21'] = 'https://a1.p2/f2.zip'
  115. CONFIG_DATA['base21'] = 'f2.zip'
  116. CONFIG_DATA['file21'] = 'arch1/platform2/file2.zip'
  117. CONFIG_DATA['origin22'] = 'https://a2.p2/f4.zip'
  118. CONFIG_DATA['base22'] = 'f4.zip'
  119. CONFIG_DATA['file22'] = 'arch2/platform2/file4.zip'
  120. CONFIG_DATA['apk2a'] = 'p2/test1.apk'
  121. CONFIG_DATA['apk2b'] = 'p2/test2.apk'
  122. DEPS_DATA = {}
  123. DEPS_DATA['template'] = """deps = {
  124. 'src/android_webview/tools/cts_archive': {
  125. 'packages': [
  126. {
  127. 'package': '%s',
  128. 'version': '%s',
  129. },
  130. ],
  131. 'condition': 'checkout_android',
  132. 'dep_type': 'cipd',
  133. },
  134. }
  135. """
  136. DEPS_DATA['revision'] = 'ctsarchiveversion'
  137. DEPS_DATA['deps'] = DEPS_DATA['template'] % (CIPD_DATA['package'],
  138. DEPS_DATA['revision'])
  139. SUITES_DATA = {}
  140. SUITES_DATA['template'] = """{
  141. # Test suites.
  142. 'basic_suites': {
  143. 'suite1': {
  144. 'webview_cts_tests': {
  145. 'swarming': {
  146. 'shards': 2,
  147. 'cipd_packages': [
  148. {
  149. "cipd_package": 'chromium/android_webview/tools/cts_archive',
  150. 'location': 'android_webview/tools/cts_archive',
  151. 'revision': '%s',
  152. }
  153. ]
  154. },
  155. },
  156. },
  157. 'suite2': {
  158. 'webview_cts_tests': {
  159. 'swarming': {
  160. 'shards': 2,
  161. 'cipd_packages': [
  162. {
  163. "cipd_package": 'chromium/android_webview/tools/cts_archive',
  164. 'location': 'android_webview/tools/cts_archive',
  165. 'revision': '%s',
  166. }
  167. ]
  168. },
  169. },
  170. },
  171. }
  172. }"""
  173. SUITES_DATA['pyl'] = SUITES_DATA['template'] % (DEPS_DATA['revision'],
  174. DEPS_DATA['revision'])
  175. GENERATE_BUILDBOT_JSON = os.path.join('testing', 'buildbot',
  176. 'generate_buildbot_json.py')
  177. _CIPD_REFERRERS = [
  178. 'DEPS', os.path.join('testing', 'buildbot', 'test_suites.pyl')
  179. ]
  180. # Used by check_tempdir.
  181. with tempfile.NamedTemporaryFile() as _f:
  182. _TEMP_DIR = os.path.dirname(_f.name) + os.path.sep
  183. class FakeCIPD:
  184. """Fake CIPD service that supports create and ensure operations."""
  185. _ensure_regex = r'\$ParanoidMode CheckIntegrity[\n\r]+' \
  186. r'@Subdir ([\w/-]+)[\n\r]+' \
  187. r'([\w/-]+) ([\w:]+)[\n\r]*'
  188. _package_json = """{
  189. "result": {
  190. "package": "%s",
  191. "instance_id": "%s"
  192. }
  193. }"""
  194. def __init__(self):
  195. self._yaml = {}
  196. self._fake_version = 0
  197. self._latest_version = {}
  198. def add_package(self, package_def, version):
  199. """Adds a version, which then becomes available for ensure operations.
  200. Args:
  201. package_def: path to package definition in cipd yaml format. The
  202. contents of each file will be set to the file name string.
  203. version: cipd version
  204. Returns:
  205. json string with same format as that of cipd ensure -json-output
  206. """
  207. with open(package_def) as def_file:
  208. yaml_dict = cts_utils.CTSCIPDYaml.parse(def_file.readlines())
  209. package = yaml_dict['package']
  210. if package not in self._yaml:
  211. self._yaml[package] = {}
  212. if version in self._yaml[package]:
  213. raise Exception('Attempting to add existing version: ' + version)
  214. self._yaml[package][version] = {}
  215. self._yaml[package][version]['yaml'] = yaml_dict
  216. self._latest_version[package] = version
  217. return self._package_json % (yaml_dict['package'], version)
  218. def get_package(self, package, version):
  219. """Gets the yaml dict of the package at version
  220. Args:
  221. package: name of cipd package
  222. version: version of cipd package
  223. Returns:
  224. Dictionary of the package in cipd yaml format
  225. """
  226. return self._yaml[package][version]['yaml']
  227. def get_latest_version(self, package):
  228. return self._latest_version.get(package)
  229. def create(self, package_def, output=None):
  230. """Implements cipd create -pkg-def <pakcage_def> [-json-output <path>]
  231. Args:
  232. package_def: path to package definition in cipd yaml format. The
  233. contents of each file will be set to the file name string.
  234. output: output file to write json formatted result
  235. Returns:
  236. json string with same format as that of cipd ensure -json-output
  237. """
  238. version = 'fake_version_' + str(self._fake_version)
  239. json_result = self.add_package(package_def, version)
  240. self._fake_version += 1
  241. if output:
  242. writefile(json_result, output)
  243. return version
  244. def ensure(self, ensure_root, ensure_file):
  245. """Implements cipd ensure -root <ensure_root> -ensure-file <ensure_file>
  246. Args:
  247. ensure_root: Base directory to copy files to
  248. ensure_file: Path to the cipd ensure file specifying the package version
  249. Raises:
  250. Exception if package and/or version was not previously added
  251. or if ensure file format is not as expected.
  252. """
  253. ensure_contents = readfile(ensure_file)
  254. match = re.match(self._ensure_regex, ensure_contents)
  255. if match:
  256. subdir = match.group(1)
  257. package = match.group(2)
  258. version = match.group(3)
  259. if package not in self._yaml:
  260. raise Exception('Package not found: ' + package)
  261. if version not in self._yaml[package]:
  262. raise Exception('Version not found: ' + version)
  263. else:
  264. raise Exception('Ensure file not recognized: ' + ensure_contents)
  265. for file_name in [e['file'] for e in \
  266. self._yaml[package][version]['yaml']['data']]:
  267. writefile(file_name,
  268. os.path.join(os.path.abspath(ensure_root), subdir, file_name))
  269. class FakeRunCmd:
  270. """Fake RunCmd that can perform cipd and cp operstions."""
  271. def __init__(self, cipd=None):
  272. self._cipd = cipd
  273. def run_cmd(self, args):
  274. """Implement devil.utils.cmd_helper.RunCmd.
  275. This doesn't implement cwd kwarg since it's not used by cts_utils
  276. Args:
  277. args: list of args
  278. """
  279. if (len(args) == 6 and args[:3] == ['cipd', 'ensure', '-root']
  280. and args[4] == '-ensure-file'):
  281. # cipd ensure -root <root> -ensure-file <file>
  282. check_tempdir(os.path.dirname(args[3]))
  283. self._cipd.ensure(args[3], args[5])
  284. elif (len(args) == 6 and args[:3] == ['cipd', 'create', '-pkg-def']
  285. and args[4] == '-json-output'):
  286. # cipd create -pkg-def <def file> -json-output <output file>
  287. check_tempdir(os.path.dirname(args[5]))
  288. self._cipd.create(args[3], args[5])
  289. elif len(args) == 4 and args[:2] == ['cp', '--reflink=never']:
  290. # cp --reflink=never <src> <dest>
  291. check_tempdir(os.path.dirname(args[3]))
  292. shutil.copyfile(args[2], args[3])
  293. elif len(args) == 3 and args[0] == 'cp':
  294. # cp <src> <dest>
  295. check_tempdir(os.path.dirname(args[2]))
  296. shutil.copyfile(args[1], args[2])
  297. else:
  298. raise Exception('Unknown cmd: ' + str(args))
  299. class CTSUtilsTest(unittest.TestCase):
  300. """Unittests for the cts_utils.py."""
  301. @unittest.skipIf(os.name == "nt", "Opening NamedTemporaryFile by name "
  302. "doesn't work in Windows.")
  303. def testCTSCIPDYamlSanity(self):
  304. yaml_data = cts_utils.CTSCIPDYaml(cts_utils.CIPD_PATH)
  305. self.assertTrue(yaml_data.get_package())
  306. self.assertTrue(yaml_data.get_files())
  307. with tempfile.NamedTemporaryFile('w+t') as outputFile:
  308. yaml_data.write(outputFile.name)
  309. with open(cts_utils.CIPD_PATH) as cipdFile:
  310. self.assertEqual(cipdFile.readlines(), outputFile.readlines())
  311. @unittest.skipIf(os.name == "nt", "Opening NamedTemporaryFile by name "
  312. "doesn't work in Windows.")
  313. def testCTSCIPDYamlOperations(self):
  314. with tempfile.NamedTemporaryFile('w+t') as yamlFile:
  315. yamlFile.writelines(CIPD_DATA['yaml'])
  316. yamlFile.flush()
  317. yaml_data = cts_utils.CTSCIPDYaml(yamlFile.name)
  318. self.assertEqual(CIPD_DATA['package'], yaml_data.get_package())
  319. self.assertEqual([
  320. CIPD_DATA['file1'], CIPD_DATA['file2'], CIPD_DATA['file3'],
  321. CIPD_DATA['file4']
  322. ], yaml_data.get_files())
  323. yaml_data.append_file('arch2/platform3/file5.zip')
  324. self.assertEqual([
  325. CIPD_DATA['file1'], CIPD_DATA['file2'], CIPD_DATA['file3'],
  326. CIPD_DATA['file4']
  327. ] + ['arch2/platform3/file5.zip'], yaml_data.get_files())
  328. yaml_data.remove_file(CIPD_DATA['file1'])
  329. self.assertEqual([
  330. CIPD_DATA['file2'], CIPD_DATA['file3'], CIPD_DATA['file4'],
  331. 'arch2/platform3/file5.zip'
  332. ], yaml_data.get_files())
  333. with tempfile.NamedTemporaryFile() as yamlFile:
  334. yaml_data.write(yamlFile.name)
  335. new_yaml_contents = readfile(yamlFile.name)
  336. self.assertEqual(
  337. CIPD_DATA['template'] %
  338. (CIPD_DATA['package'], CIPD_DATA['file2'], CIPD_DATA['file3'],
  339. CIPD_DATA['file4'], 'arch2/platform3/file5.zip'), new_yaml_contents)
  340. @patch('devil.utils.cmd_helper.RunCmd')
  341. @unittest.skipIf(os.name == "nt", "Opening NamedTemporaryFile by name "
  342. "doesn't work in Windows.")
  343. def testCTSCIPDDownload(self, run_mock):
  344. fake_cipd = FakeCIPD()
  345. fake_run_cmd = FakeRunCmd(cipd=fake_cipd)
  346. run_mock.side_effect = fake_run_cmd.run_cmd
  347. with tempfile.NamedTemporaryFile('w+t') as yamlFile,\
  348. tempfile_ext.NamedTemporaryDirectory() as tempDir:
  349. yamlFile.writelines(CIPD_DATA['yaml'])
  350. yamlFile.flush()
  351. fake_version = fake_cipd.create(yamlFile.name)
  352. archive = cts_utils.CTSCIPDYaml(yamlFile.name)
  353. cts_utils.cipd_download(archive, fake_version, tempDir)
  354. self.assertEqual(CIPD_DATA['file1'],
  355. readfile(os.path.join(tempDir, CIPD_DATA['file1'])))
  356. self.assertEqual(CIPD_DATA['file2'],
  357. readfile(os.path.join(tempDir, CIPD_DATA['file2'])))
  358. def testCTSConfigSanity(self):
  359. cts_config = cts_utils.CTSConfig()
  360. platforms = cts_config.get_platforms()
  361. self.assertTrue(platforms)
  362. platform = platforms[0]
  363. archs = cts_config.get_archs(platform)
  364. self.assertTrue(archs)
  365. self.assertTrue(cts_config.get_cipd_zip(platform, archs[0]))
  366. self.assertTrue(cts_config.get_origin(platform, archs[0]))
  367. self.assertTrue(cts_config.get_apks(platform))
  368. @unittest.skipIf(os.name == "nt", "Opening NamedTemporaryFile by name "
  369. "doesn't work in Windows.")
  370. def testCTSConfig(self):
  371. with tempfile.NamedTemporaryFile('w+t') as configFile:
  372. configFile.writelines(CONFIG_DATA['json'])
  373. configFile.flush()
  374. cts_config = cts_utils.CTSConfig(configFile.name)
  375. self.assertEqual(['platform1', 'platform2'], cts_config.get_platforms())
  376. self.assertEqual(['arch1', 'arch2'], cts_config.get_archs('platform1'))
  377. self.assertEqual(['arch1', 'arch2'], cts_config.get_archs('platform2'))
  378. self.assertEqual('arch1/platform1/file1.zip',
  379. cts_config.get_cipd_zip('platform1', 'arch1'))
  380. self.assertEqual('arch2/platform1/file3.zip',
  381. cts_config.get_cipd_zip('platform1', 'arch2'))
  382. self.assertEqual('arch1/platform2/file2.zip',
  383. cts_config.get_cipd_zip('platform2', 'arch1'))
  384. self.assertEqual('arch2/platform2/file4.zip',
  385. cts_config.get_cipd_zip('platform2', 'arch2'))
  386. self.assertEqual('https://a1.p1/f1.zip',
  387. cts_config.get_origin('platform1', 'arch1'))
  388. self.assertEqual('https://a2.p1/f3.zip',
  389. cts_config.get_origin('platform1', 'arch2'))
  390. self.assertEqual('https://a1.p2/f2.zip',
  391. cts_config.get_origin('platform2', 'arch1'))
  392. self.assertEqual('https://a2.p2/f4.zip',
  393. cts_config.get_origin('platform2', 'arch2'))
  394. self.assertTrue(['p1/test.apk'], cts_config.get_apks('platform1'))
  395. self.assertTrue(['p2/test1.apk', 'p2/test2.apk'],
  396. cts_config.get_apks('platform2'))
  397. self.assertTrue([
  398. 'p2/additional_apk_a_1.apk', 'p2/additional_apk_b_1.apk',
  399. 'p2/additional_apk_b_2.apk'
  400. ], cts_config.get_additional_apks('platform2'))
  401. @unittest.skipIf(os.name == "nt", "This fails on Windows, probably because "
  402. "the temporary directory is not empty when it gets deleted.")
  403. def testFilterZip(self):
  404. with tempfile_ext.NamedTemporaryDirectory() as workDir,\
  405. cts_utils.chdir(workDir):
  406. writefile('abc', 'a/b/one.apk')
  407. writefile('def', 'a/b/two.apk')
  408. writefile('ghi', 'a/b/three.apk')
  409. movetozip(['a/b/one.apk', 'a/b/two.apk', 'a/b/three.apk'],
  410. 'downloaded.zip')
  411. cts_utils.filterzip('downloaded.zip', ['a/b/one.apk', 'a/b/two.apk'],
  412. 'filtered.zip')
  413. zf = zipfile.ZipFile('filtered.zip', 'r')
  414. self.assertEqual(2, len(zf.namelist()))
  415. self.assertEqual(b'abc', zf.read('a/b/one.apk'))
  416. self.assertEqual(b'def', zf.read('a/b/two.apk'))
  417. @patch('cts_utils.filterzip')
  418. @unittest.skipIf(os.name == "nt", "Opening NamedTemporaryFile by name "
  419. "doesn't work in Windows.")
  420. # pylint: disable=no-self-use
  421. def testFilterCTS(self, filterzip_mock):
  422. with tempfile.NamedTemporaryFile('w+t') as configFile:
  423. configFile.writelines(CONFIG_DATA['json'])
  424. configFile.flush()
  425. cts_config = cts_utils.CTSConfig(configFile.name)
  426. cts_utils.filter_cts_file(cts_config, CONFIG_DATA['base11'], '/filtered')
  427. filterzip_mock.assert_called_with(
  428. CONFIG_DATA['base11'], [CONFIG_DATA['apk1']],
  429. os.path.join('/filtered', CONFIG_DATA['base11']))
  430. @patch('devil.utils.cmd_helper.RunCmd')
  431. @unittest.skipIf(os.name == "nt", "Opening NamedTemporaryFile by name "
  432. "doesn't work in Windows.")
  433. def testUpdateCIPDPackage(self, run_mock):
  434. fake_cipd = FakeCIPD()
  435. fake_run_cmd = FakeRunCmd(cipd=fake_cipd)
  436. run_mock.side_effect = fake_run_cmd.run_cmd
  437. with tempfile_ext.NamedTemporaryDirectory() as tempDir,\
  438. cts_utils.chdir(tempDir):
  439. writefile(CIPD_DATA['yaml'], 'cipd.yaml')
  440. version = cts_utils.update_cipd_package('cipd.yaml')
  441. uploaded = fake_cipd.get_package(CIPD_DATA['package'], version)
  442. self.assertEqual(CIPD_DATA['package'], uploaded['package'])
  443. uploaded_files = [e['file'] for e in uploaded['data']]
  444. self.assertEqual(4, len(uploaded_files))
  445. for i in range(1, 5):
  446. self.assertTrue(CIPD_DATA['file' + str(i)] in uploaded_files)
  447. def testChromiumRepoHelper(self):
  448. with tempfile_ext.NamedTemporaryDirectory() as tempDir,\
  449. cts_utils.chdir(tempDir):
  450. setup_fake_repo('.')
  451. helper = cts_utils.ChromiumRepoHelper(root_dir='.')
  452. self.assertEqual(DEPS_DATA['revision'], helper.get_cipd_dependency_rev())
  453. self.assertEqual(os.path.join(tempDir, 'a', 'b'), helper.rebase('a', 'b'))
  454. helper.update_cts_cipd_rev('newversion')
  455. self.assertEqual('newversion', helper.get_cipd_dependency_rev())
  456. expected_deps = DEPS_DATA['template'] % (CIPD_DATA['package'],
  457. 'newversion')
  458. self.assertEqual(expected_deps, readfile(_CIPD_REFERRERS[0]))
  459. expected_suites = SUITES_DATA['template'] % ('newversion', 'newversion')
  460. self.assertEqual(expected_suites, readfile(_CIPD_REFERRERS[1]))
  461. writefile('#deps not referring to cts cipd', _CIPD_REFERRERS[0])
  462. with self.assertRaises(Exception):
  463. helper.update_cts_cipd_rev('anothernewversion')
  464. @patch('urllib.urlretrieve' if six.PY2 else 'urllib.request.urlretrieve')
  465. @patch('os.makedirs')
  466. # pylint: disable=no-self-use
  467. def testDownload(self, mock_makedirs, mock_retrieve):
  468. t1 = cts_utils.download('http://www.download.com/file1.zip',
  469. '/download_dir/file1.zip')
  470. t2 = cts_utils.download('http://www.download.com/file2.zip',
  471. '/download_dir/file2.zip')
  472. t1.join()
  473. t2.join()
  474. mock_makedirs.assert_called_with('/download_dir')
  475. mock_retrieve.assert_any_call('http://www.download.com/file1.zip',
  476. '/download_dir/file1.zip')
  477. mock_retrieve.assert_any_call('http://www.download.com/file2.zip',
  478. '/download_dir/file2.zip')
  479. def setup_fake_repo(repoRoot):
  480. """Populates various files needed for testing cts_utils.
  481. Args:
  482. repo_root: Root of the fake repo under which to write config files
  483. """
  484. with cts_utils.chdir(repoRoot):
  485. writefile(DEPS_DATA['deps'], cts_utils.DEPS_FILE)
  486. writefile(CONFIG_DATA['json'],
  487. os.path.join(cts_utils.TOOLS_DIR, cts_utils.CONFIG_FILE))
  488. writefile(CIPD_DATA['yaml'],
  489. os.path.join(cts_utils.TOOLS_DIR, cts_utils.CIPD_FILE))
  490. writefile(SUITES_DATA['pyl'], cts_utils.TEST_SUITES_FILE)
  491. def readfile(fpath):
  492. """Returns contents of file at fpath."""
  493. with open(fpath) as f:
  494. return f.read()
  495. def writefile(contents, path):
  496. """Writes contents to file at path."""
  497. dir_path = os.path.dirname(os.path.abspath(path))
  498. if not os.path.isdir(dir_path):
  499. os.makedirs(os.path.dirname(path))
  500. with open(path, 'w') as f:
  501. f.write(contents)
  502. def movetozip(fileList, outputPath):
  503. """Move files in fileList to zip file at outputPath"""
  504. with zipfile.ZipFile(outputPath, 'a') as zf:
  505. for f in fileList:
  506. zf.write(f)
  507. os.remove(f)
  508. def check_tempdir(path):
  509. """Check if directory at path is under tempdir.
  510. Args:
  511. path: path of directory to check
  512. Raises:
  513. AssertionError if directory is not under tempdir.
  514. """
  515. abs_path = os.path.abspath(path) + os.path.sep
  516. if abs_path[:len(_TEMP_DIR)] != _TEMP_DIR:
  517. raise AssertionError(
  518. '"%s" is not under tempdir "%s".' % (abs_path, _TEMP_DIR))
  519. if __name__ == '__main__':
  520. unittest.main()