Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def mocked_new_sources(sources=None):
if not Path(sources).is_file():
raise RuntimeError("archive does not exist")
flexmock(FedPKG, new_sources=mocked_new_sources)
flexmock(PackitAPI, init_kerberos_ticket=lambda: None)
flexmock(
DistGit,
push_to_fork=lambda *args, **kwargs: None,
is_archive_in_lookaside_cache=lambda archive_path: False,
upload_to_lookaside_cache=lambda path: None,
download_upstream_archive=lambda: "the-archive",
)
flexmock(
PackitAPI,
push_and_create_pr=lambda pr_title, pr_description, dist_git_branch: None,
)
pc = get_local_package_config(str(upstream_path))
up_lp = LocalProject(working_dir=str(upstream_path))
c = get_test_config()
api = PackitAPI(c, pc, up_lp)
api._dg = DistGit(c, pc)
api._dg._local_project = LocalProject(working_dir=dist_git_path)
with cwd(upstream_path):
api.sync_release(
"master",
use_local_content=False,
version="179",
force_new_sources=False,
def test_srpm_snapd(snapd):
pc = get_local_package_config(str(snapd))
up_lp = LocalProject(working_dir=str(snapd))
c = get_test_config()
api = PackitAPI(c, pc, up_lp)
with cwd(snapd):
path = api.create_srpm()
assert path.exists()
build_srpm(path)
def test_run_copr_build(upstream_n_distgit, copr_project, copr_build, test_copr_client):
u, d = upstream_n_distgit
with cwd(u):
c = get_test_config()
pc = get_local_package_config(str(u))
pc.upstream_project_url = str(u)
pc.downstream_project_url = str(d)
up_lp = LocalProject(path_or_url=str(u))
api = PackitAPI(c, pc, up_lp)
flexmock(Client).should_receive("create_from_config_file").and_return(
test_copr_client
)
# invalid owner + project
with pytest.raises(PackitInvalidConfigException):
api.run_copr_build("not-packit", "dummy", [])
# with project chroots update
flexmock(proxies.project.ProjectProxy).should_receive("get").and_return(
copr_project
).once()
flexmock(proxies.project.ProjectProxy).should_receive("edit").once()
flexmock(proxies.build.BuildProxy).should_receive(
def setUp(self):
super().setUp()
self.api = PackitAPI(
config=self.conf, package_config=self.pc, upstream_local_project=self.lp
)
self.api._up = self.upstream
self.api._dg = self.dg
# Do not upload package, because no credentials given in CI
flexmock(self.api).should_receive("_handle_sources").and_return(None)
flexmock(self.api.dg).should_receive("push").and_return(None)
flexmock(git.HEAD).should_receive("commit").and_return("hash-of-some-commit")
self.set_git_user()
async def status_main(status: Status) -> List:
"""
Schedule repository data retrieval calls concurrently.
:param status: status of the package
:return: awaitable tasks
"""
res = await asyncio.gather(
PackitAPI.status_get_downstream_prs(status),
PackitAPI.status_get_dg_versions(status),
PackitAPI.status_get_up_releases(status),
PackitAPI.status_get_koji_builds(status),
PackitAPI.status_get_copr_builds(status),
PackitAPI.status_get_updates(status),
)
return res
def validate_config(path_or_url: LocalProject):
"""
Validate PackageConfig validation.
\b
- checks missing values
- checks incorrect types
PATH_OR_URL argument is a local path or a URL to a git repository with packit configuration file
"""
# we use PackageConfig.load_from_dict for the validation, hence we don't parse it here
output = PackitAPI.validate_package_config(Path(path_or_url.working_dir))
logger.info(output)
# TODO: print more if config.debug
self,
package_config: PackageConfig,
version: Optional[str],
dist_git_branch: str,
upstream_local_project: LocalProject,
):
"""
Sync the upstream release to the distgit pull-request.
:param package_config: PackageConfig
:param version: not used now, str
:param dist_git_branch: str
:param upstream_local_project: LocalProject instance
"""
logger.info("syncing the upstream code to downstream")
packit_api = PackitAPI(
config=self.config,
package_config=package_config,
upstream_local_project=upstream_local_project,
)
packit_api.sync_release(dist_git_branch=dist_git_branch, version=version)
async def status_main(status: Status) -> List:
"""
Schedule repository data retrieval calls concurrently.
:param status: status of the package
:return: awaitable tasks
"""
res = await asyncio.gather(
PackitAPI.status_get_downstream_prs(status),
PackitAPI.status_get_dg_versions(status),
PackitAPI.status_get_up_releases(status),
PackitAPI.status_get_koji_builds(status),
PackitAPI.status_get_copr_builds(status),
PackitAPI.status_get_updates(status),
)
return res
async def status_main(status: Status) -> List:
"""
Schedule repository data retrieval calls concurrently.
:param status: status of the package
:return: awaitable tasks
"""
res = await asyncio.gather(
PackitAPI.status_get_downstream_prs(status),
PackitAPI.status_get_dg_versions(status),
PackitAPI.status_get_up_releases(status),
PackitAPI.status_get_koji_builds(status),
PackitAPI.status_get_copr_builds(status),
PackitAPI.status_get_updates(status),
)
return res
def sync_upstream_pull_request(
self,
package_config: PackageConfig,
pr_id: int,
dist_git_branch: str,
upstream_local_project: LocalProject,
):
logger.info("syncing the upstream code to downstream")
packit_api = PackitAPI(
config=self.config,
package_config=package_config,
upstream_local_project=upstream_local_project,
)
packit_api.sync_pr(pr_id=pr_id, dist_git_branch=dist_git_branch)