Skip to content

Commit

Permalink
chore: rename _getOCIsForRepo as _getTagsForRepo()
Browse files Browse the repository at this point in the history
  • Loading branch information
@jdstrand
jdstrand committed Aug 18, 2023
1 parent 3fd9bc9 commit abc66e6
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 29 deletions.
26 changes: 13 additions & 13 deletions cvelib/dso.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -202,24 +202,20 @@ def getDigestForImage(self, repo_full: str) -> str:
return ""

def parseImageDigest(self, digest: str) -> Tuple[str, str, str]:
"""Parse the image digest into a (namespace, repo, sha256) tuple"""
"""Parse the image digest into a (namespace (ignored), repo, sha256) tuple"""
if "@sha256:" not in digest:
error("Malformed digest '%s' (does not contain '@sha256:')" % digest)
return ("", "", "")
elif digest.count("@") != 1:
error("Malformed digest '%s' (should have 1 '@')" % digest)
return ("", "", "")

sha256: str = ""
repo: str
sha256: str
repo, sha256 = digest.split("@")

return ("", repo, sha256)

def getOCIsForNamespace(self, _: str) -> List[Tuple[str, int]]: # pragma: nocover
"""Obtain the list of DockerDSO repos for the specified namespace"""
# dso doesn't have a concept of namespaces
raise NotImplementedError

def fetchScanReport(
self,
repo_full: str,
Expand Down Expand Up @@ -270,6 +266,10 @@ def fetchScanReport(

return ocis, ""

def getOCIsForNamespace(self, _: str) -> List[Tuple[str, int]]: # pragma: nocover
# dso doesn't have a concept of namespaces
raise NotImplementedError

def getReposForNamespace(self, _: str) -> List[str]: # pragma: nocover
# dso doesn't have a concept of repos within namespaces
raise NotImplementedError
Expand Down Expand Up @@ -752,7 +752,7 @@ def _getListEDN(namespace: str, days: int = 365) -> Dict:
# "x-atomist-correlation-id": "81e2aee7-13d1-4097-93aa-90841e5bd43b"
# }
# }
def _getOCIsForRepo(repo_name: str) -> List[Tuple[str, int]]:
def _getTagsForRepo(repo_name: str) -> List[Tuple[str, int]]:
"""Obtain the list of DockerDSO tags for the specified repo"""
if ":" in repo_name or "@" in repo_name or "/" in repo_name:
error("Please use REPO (without :TAG or @sha256:SHA256)")
Expand Down Expand Up @@ -810,10 +810,10 @@ def main_dso_dump_reports():
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=textwrap.dedent(
"""\
dso-dump-reports pulls all the latest security reports for OCI images in
REPO and outputs them to:
dso-dump-reports pulls all the latest security reports for the tagged images in
the REPO and outputs them to:
/path/to/reports/YY/MM/DD/dso/REPO/TAG/SHA256.json
/path/to/reports/YY/MM/DD/dso/REPO/SHA256.json
Eg, to pull all dso security scan reports for org 'foo':
Expand Down Expand Up @@ -843,7 +843,7 @@ def main_dso_dump_reports():
sr = DockerDSOSecurityReportNew()

# Find latest digest for all images
oci_names: List[Tuple[str, int]] = _getOCIsForRepo(args.name)
oci_names: List[Tuple[str, int]] = _getTagsForRepo(args.name)
if len(oci_names) == 0:
error("Could not enumerate any OCI image names")
return # for tests
Expand Down Expand Up @@ -872,7 +872,7 @@ def main_dso_dump_reports():
# dso doesn't have dates or times in the security report, so we will
# store them in a folder under today's date. Since the report path comes
# from the date the report was fetched, we'll first search for the report
# by the dso/TAG/SHA256.json to see if we previously downloaded it.
# by the dso/REPO/SHA256.json to see if we previously downloaded it.

# gather a list of potentially matching filenames
json_files: Dict[str, str] = {}
Expand Down
32 changes: 16 additions & 16 deletions tests/test_dso.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -376,8 +376,8 @@ def test__getListEDN(self, mock_post, mock_ednLoadAsDict):
# Note, these are listed in reverse order ot the arguments to test_...
@mock.patch("cvelib.dso.ednLoadAsDict")
@mock.patch("requests.post")
def test__getOCIsForRepo(self, mock_post, mock_ednLoadAsDict):
"""Test _getOCIsForRepo()"""
def test__getTagsForRepo(self, mock_post, mock_ednLoadAsDict):
"""Test _getTagsForRepo()"""
mock_post.return_value = self._mock_response_for_dso(content="edn-doc")
mock_ednLoadAsDict.return_value = {
"docker-repository-tags": {
Expand All @@ -403,7 +403,7 @@ def test__getOCIsForRepo(self, mock_post, mock_ednLoadAsDict):
"x-atomist-correlation-id": "81e2aee7-13d1-4097-93aa-90841e5bd43b"
},
}
res = cvelib.dso._getOCIsForRepo("valid-repo")
res = cvelib.dso._getTagsForRepo("valid-repo")
self.assertEqual(1, len(res))
self.assertEqual("1.0-valid-name", res[0][0])

Expand Down Expand Up @@ -442,15 +442,15 @@ def test__getOCIsForRepo(self, mock_post, mock_ednLoadAsDict):
"x-atomist-correlation-id": "81e2aee7-13d1-4097-93aa-90841e5bd43b"
},
}
res = cvelib.dso._getOCIsForRepo("valid-repo")
res = cvelib.dso._getTagsForRepo("valid-repo")
self.assertEqual(1, len(res))
self.assertEqual(0, res[0][1])

# empty
mock_post.return_value = self._mock_response_for_dso(content="edn-doc")
mock_ednLoadAsDict.return_value = {}
with tests.testutil.capturedOutput() as (output, error):
res = cvelib.dso._getOCIsForRepo("valid-repo")
res = cvelib.dso._getTagsForRepo("valid-repo")
self.assertEqual("", output.getvalue().strip())
self.assertTrue(
"Could not find 'docker-repository-tags' as dict in response"
Expand All @@ -471,7 +471,7 @@ def test__getOCIsForRepo(self, mock_post, mock_ednLoadAsDict):
},
}
with tests.testutil.capturedOutput() as (output, error):
res = cvelib.dso._getOCIsForRepo("valid-repo")
res = cvelib.dso._getTagsForRepo("valid-repo")
self.assertEqual("", output.getvalue().strip())
self.assertTrue(
"Could not find 'image' in response for image" in error.getvalue().strip()
Expand All @@ -488,7 +488,7 @@ def test__getOCIsForRepo(self, mock_post, mock_ednLoadAsDict):
),
):
with tests.testutil.capturedOutput() as (output, error):
res = cvelib.dso._getOCIsForRepo("valid-repo:dont-use-tag")
res = cvelib.dso._getTagsForRepo("valid-repo:dont-use-tag")
self.assertEqual("", output.getvalue().strip())
self.assertTrue(
"Please use REPO (without :TAG or @sha256:SHA256)"
Expand Down Expand Up @@ -919,18 +919,18 @@ def test_fetchScanReport(self, mock_post, mock_fetchVulnReports):
# Note, these are listed in reverse order ot the arguments to test_...
@mock.patch("cvelib.dso.DockerDSOSecurityReportNew.fetchScanReport")
@mock.patch("cvelib.dso.DockerDSOSecurityReportNew.getDigestForImage")
@mock.patch("cvelib.dso._getOCIsForRepo")
@mock.patch("cvelib.dso._getTagsForRepo")
def test_main_dso_dump_reports(
self,
mock__getOCIsForRepo,
mock__getTagsForRepo,
mock_getDigestForImage,
mock_fetchScanReport,
):
"""Test test_main_dso_dump_reports()"""
self.tmpdir = tempfile.mkdtemp(prefix="sedg-")
os.environ["SEDG_EXPERIMENTAL"] = "1"

mock__getOCIsForRepo.return_value = [("valid-name", 1684472852)]
mock__getTagsForRepo.return_value = [("valid-name", 1684472852)]
mock_getDigestForImage.return_value = "valid-name@sha256:deadbeef"
mock_fetchScanReport.return_value = (
[],
Expand Down Expand Up @@ -1009,10 +1009,10 @@ def test_main_dso_dump_reports(
# Note, these are listed in reverse order ot the arguments to test_...
@mock.patch("cvelib.dso.DockerDSOSecurityReportNew.fetchScanReport")
@mock.patch("cvelib.dso.DockerDSOSecurityReportNew.getDigestForImage")
@mock.patch("cvelib.dso._getOCIsForRepo")
@mock.patch("cvelib.dso._getTagsForRepo")
def test_main_dso_dump_reports_bad(
self,
mock__getOCIsForRepo,
mock__getTagsForRepo,
mock_getDigestForImage,
mock_fetchScanReport,
):
Expand All @@ -1021,7 +1021,7 @@ def test_main_dso_dump_reports_bad(
os.environ["SEDG_EXPERIMENTAL"] = "1"

# no image names
mock__getOCIsForRepo.return_value = []
mock__getTagsForRepo.return_value = []
with mock.patch.object(
cvelib.common.error,
"__defaults__",
Expand All @@ -1048,7 +1048,7 @@ def test_main_dso_dump_reports_bad(
)

# no digests
mock__getOCIsForRepo.return_value = [("valid-name", 1684472852)]
mock__getTagsForRepo.return_value = [("valid-name", 1684472852)]
mock_getDigestForImage.return_value = ""
with mock.patch.object(
cvelib.common.error,
Expand Down Expand Up @@ -1078,7 +1078,7 @@ def test_main_dso_dump_reports_bad(
"Could not find any OCI image digests" in error.getvalue().strip(),
)

mock__getOCIsForRepo.return_value = [("valid-name", 1684472852)]
mock__getTagsForRepo.return_value = [("valid-name", 1684472852)]
mock_getDigestForImage.return_value = "valid-name@sha256:deadbeef"
mock_fetchScanReport.return_value = [], ""
with mock.patch.object(
Expand All @@ -1105,7 +1105,7 @@ def test_main_dso_dump_reports_bad(
self.assertTrue("No new security reports" in error.getvalue().strip())

# unsupported scan status
mock__getOCIsForRepo.return_value = [("valid-name", 1684472852)]
mock__getTagsForRepo.return_value = [("valid-name", 1684472852)]
mock_getDigestForImage.return_value = "valid-name@sha256:deadbeef"
mock_fetchScanReport.return_value = ([], '{"data": null}')
with mock.patch.object(
Expand Down

0 comments on commit abc66e6

Please sign in to comment.