-
Notifications
You must be signed in to change notification settings - Fork 134
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
RUMM-1744 Add
DirectoryMatcher type to distribution tool
to simplify assertion of GH asset content
- Loading branch information
Showing
3 changed files
with
213 additions
and
40 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| # ----------------------------------------------------------- | ||
| # Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0. | ||
| # This product includes software developed at Datadog (https://www.datadoghq.com/). | ||
| # Copyright 2019-2020 Datadog, Inc. | ||
| # ----------------------------------------------------------- | ||
|
|
||
| import os | ||
| import glob | ||
|
|
||
|
|
||
| class DirectoryMatcherException(Exception): | ||
| pass | ||
|
|
||
|
|
||
| class DirectoryMatcher: | ||
| path: str | ||
|
|
||
| def __init__(self, path: str): | ||
| if os.path.exists(path): | ||
| self.path = path | ||
| else: | ||
| raise DirectoryMatcherException(f'Directory does not exist: {path}') | ||
|
|
||
| def assert_number_of_files(self, expected_count: int): | ||
| actual_count = len(os.listdir(self.path)) | ||
| if expected_count != actual_count: | ||
| raise DirectoryMatcherException(f'Expected {expected_count} files in "{self.path}", but ' | ||
| f'found {actual_count} instead.') | ||
|
|
||
| def assert_it_has_file(self, file_path: str): | ||
| search_path = os.path.join(self.path, file_path) | ||
| result = list(glob.iglob(search_path, recursive=True)) | ||
|
|
||
| # if not os.path.exists(os.path.join(self.path, file_path)): | ||
| if not result: | ||
| raise DirectoryMatcherException(f'Expected "{self.path}" to include {file_path}, but it is missing.') | ||
|
|
||
| def assert_it_has_files(self, file_paths: [str]): | ||
| for file_path in file_paths: | ||
| self.assert_it_has_file(file_path) | ||
|
|
||
| def get(self, file: str) -> 'DirectoryMatcher': | ||
| return DirectoryMatcher(path=os.path.join(self.path, file)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| # ----------------------------------------------------------- | ||
| # Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0. | ||
| # This product includes software developed at Datadog (https://www.datadoghq.com/). | ||
| # Copyright 2019-2020 Datadog, Inc. | ||
| # ----------------------------------------------------------- | ||
|
|
||
|
|
||
| import unittest | ||
| import os | ||
| from tempfile import TemporaryDirectory | ||
| from src.directory_matcher import DirectoryMatcher, DirectoryMatcherException | ||
|
|
||
|
|
||
| class DirectoryMatcherTestCase(unittest.TestCase): | ||
| def test_initializing(self): | ||
| with TemporaryDirectory() as tmp_dir: | ||
| self.assertEqual(DirectoryMatcher(path=tmp_dir).path, tmp_dir) | ||
|
|
||
| with self.assertRaises(DirectoryMatcherException): | ||
| _ = DirectoryMatcher(path=f'{tmp_dir}/non-existing-path') | ||
|
|
||
| def test_number_of_files(self): | ||
| with TemporaryDirectory() as tmp_dir: | ||
| dm = DirectoryMatcher(path=tmp_dir) | ||
| dm.assert_number_of_files(expected_count=0) | ||
|
|
||
| os.mkdir(os.path.join(tmp_dir, '1')) | ||
| os.mkdir(os.path.join(tmp_dir, '2')) | ||
| dm.assert_number_of_files(expected_count=2) | ||
|
|
||
| with self.assertRaises(DirectoryMatcherException): | ||
| dm.assert_number_of_files(expected_count=4) | ||
|
|
||
| def test_has_file(self): | ||
| with TemporaryDirectory() as tmp_dir: | ||
| dm = DirectoryMatcher(path=tmp_dir) | ||
| os.makedirs(os.path.join(tmp_dir, '1/1A/1AA.xyz')) | ||
| os.makedirs(os.path.join(tmp_dir, '1/1A/1AB.xyz')) | ||
| os.makedirs(os.path.join(tmp_dir, '2/2A/2AA/foo.xyz')) | ||
| os.makedirs(os.path.join(tmp_dir, '2/2A/2AB/foo.xyz')) | ||
|
|
||
| dm.assert_it_has_file('1') | ||
| dm.assert_it_has_file('1/1A/1AA.xyz') | ||
| dm.assert_it_has_file('1/1A/1AB.xyz') | ||
| dm.assert_it_has_file('**/1AA.xyz') | ||
| dm.assert_it_has_file('**/1AB.xyz') | ||
| dm.assert_it_has_file('**/*.xyz') | ||
| dm.assert_it_has_file('**/2A/**/*.xyz') | ||
| dm.assert_it_has_file('2') | ||
|
|
||
| with self.assertRaises(DirectoryMatcherException): | ||
| dm.assert_it_has_file('foo') | ||
|
|
||
| with self.assertRaises(DirectoryMatcherException): | ||
| dm.assert_it_has_file('1A') | ||
|
|
||
| def test_has_files(self): | ||
| with TemporaryDirectory() as tmp_dir: | ||
| dm = DirectoryMatcher(path=tmp_dir) | ||
| os.makedirs(os.path.join(tmp_dir, '1/1A/1AA.xyz')) | ||
| os.makedirs(os.path.join(tmp_dir, '1/1A/1AB.xyz')) | ||
| os.makedirs(os.path.join(tmp_dir, '2/2A/2AA/foo.xyz')) | ||
| os.makedirs(os.path.join(tmp_dir, '2/2A/2AB/foo.xyz')) | ||
|
|
||
| dm.assert_it_has_files([ | ||
| '1', | ||
| '1/1A/1AA.xyz', | ||
| '1/1A/1AB.xyz', | ||
| '**/1AA.xyz', | ||
| '**/1AB.xyz', | ||
| '**/*.xyz', | ||
| '**/2A/**/*.xyz', | ||
| '2', | ||
| ]) | ||
|
|
||
| with self.assertRaises(DirectoryMatcherException): | ||
| dm.assert_it_has_files(file_paths=['foo', 'bar']) | ||
|
|
||
| with self.assertRaises(DirectoryMatcherException): | ||
| dm.assert_it_has_files(file_paths=['2', '**/foo']) | ||
|
|
||
| def test_get_submatcher(self): | ||
| with TemporaryDirectory() as tmp_dir: | ||
| os.makedirs(os.path.join(tmp_dir, '1/1A')) | ||
|
|
||
| dm = DirectoryMatcher(path=tmp_dir) | ||
| dm.assert_it_has_file('1') | ||
|
|
||
| dm = dm.get('1') | ||
| dm.assert_it_has_file('1A') | ||
|
|
||
| with self.assertRaises(DirectoryMatcherException): | ||
| dm.assert_it_has_file('1') |