Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_match_root_dir_pathlib(self):
"""Test root directory with `globmatch` using `pathlib`."""
from wcmatch import pathlib
self.assertFalse(glob.globmatch(pathlib.Path('markdown'), 'markdown', flags=glob.REALPATH))
self.assertTrue(
glob.globmatch(pathlib.Path('markdown'), 'markdown', flags=glob.REALPATH, root_dir=pathlib.Path('docs/src'))
)
""" Get packages paths to run lint command.
Args:
content_repo(git.Repo): Content repository object.
input(str): dir pack specified as argument.
git(bool): Perform lint and test only on changed packs.
all_packs(bool): Whether to run on all packages.
Returns:
List[Path]: Pkgs to run lint
"""
pkgs: list
if all_packs or git:
pkgs = LintManager._get_all_packages(content_dir=content_repo.working_dir)
elif not all_packs and not git and not input:
pkgs = [Path().cwd()]
else:
pkgs = []
for item in input.split(','):
is_pack = os.path.isdir(item) and os.path.exists(os.path.join(item, PACKS_PACK_META_FILE_NAME))
if is_pack:
pkgs.extend(LintManager._get_all_packages(content_dir=item))
else:
pkgs.append(Path(item))
total_found = len(pkgs)
if git:
pkgs = LintManager._filter_changed_packages(content_repo=content_repo,
pkgs=pkgs)
for pkg in pkgs:
print_v(f"Found changed package {Colors.Fg.cyan}{pkg}{Colors.reset}",
log_verbose=self._verbose)
for trial in range(2):
try:
self._docker_client.images.push(test_image_name)
logger.info(f"{log_prompt} - Image {test_image_name} pushed to repository")
break
except (requests.exceptions.ConnectionError, urllib3.exceptions.ReadTimeoutError):
logger.info(f"{log_prompt} - Unable to push image {test_image_name} to repository")
except (docker.errors.BuildError, docker.errors.APIError, Exception) as e:
logger.critical(f"{log_prompt} - Build errors occurred {e}")
errors = str(e)
else:
logger.info(f"{log_prompt} - Found existing image {test_image_name}")
for trial in range(2):
dockerfile_path = Path(self._pack_abs_dir / ".Dockerfile")
try:
logger.info(f"{log_prompt} - Copy pack dir to image {test_image_name}")
dockerfile = template.render(image=test_image_name,
copy_pack=True)
with open(file=dockerfile_path, mode="+x") as f:
f.write(str(dockerfile))
docker_image_final = self._docker_client.images.build(path=str(dockerfile_path.parent),
dockerfile=dockerfile_path.stem,
forcerm=True)
test_image_name = docker_image_final[0].short_id
break
except (docker.errors.ImageNotFound, docker.errors.APIError, urllib3.exceptions.ReadTimeoutError,
exceptions.TemplateError) as e:
logger.info(f"{log_prompt} - errors occurred when copy pack dir {e}")
if trial == 2:
""" Checks which packages had changes using git (working tree, index, diff between HEAD and master in them and should
run on Lint.
Args:
pkgs(List[Path]): pkgs to check
Returns:
List[Path]: A list of names of packages that should run.
"""
print(f"Comparing to {Colors.Fg.cyan}{content_repo.remote()}/master{Colors.reset} using branch {Colors.Fg.cyan}"
f"{content_repo.active_branch}{Colors.reset}")
staged_files = {content_repo.working_dir / Path(item.b_path).parent for item in
content_repo.active_branch.commit.tree.diff(None, paths=pkgs)}
last_common_commit = content_repo.merge_base(content_repo.active_branch.commit,
content_repo.remote().refs.master)
changed_from_master = {content_repo.working_dir / Path(item.b_path).parent for item in
content_repo.active_branch.commit.tree.diff(last_common_commit, paths=pkgs)}
all_changed = staged_files.union(changed_from_master)
pkgs_to_check = all_changed.intersection(pkgs)
return list(pkgs_to_check)
def _get_all_packages(content_dir: str) -> List[str]:
"""Gets all integration, script in packages and packs in content repo.
Returns:
list: A list of integration, script and beta_integration names.
"""
# Get packages from main content path
content_main_pkgs: set = set(Path(content_dir).glob(['Integrations/*/',
'Scripts/*/', ]))
# Get packages from packs path
packs_dir: Path = Path(content_dir) / 'Packs'
content_packs_pkgs: set = set(packs_dir.glob(['*/Integrations/*/',
'*/Scripts/*/']))
all_pkgs = content_packs_pkgs.union(content_main_pkgs)
return list(all_pkgs)
def _get_all_packages(content_dir: str) -> List[str]:
"""Gets all integration, script in packages and packs in content repo.
Returns:
list: A list of integration, script and beta_integration names.
"""
# Get packages from main content path
content_main_pkgs: set = set(Path(content_dir).glob(['Integrations/*/',
'Scripts/*/', ]))
# Get packages from packs path
packs_dir: Path = Path(content_dir) / 'Packs'
content_packs_pkgs: set = set(packs_dir.glob(['*/Integrations/*/',
'*/Scripts/*/']))
all_pkgs = content_packs_pkgs.union(content_main_pkgs)
return list(all_pkgs)