fix linting and tests

Signed-off-by: Alex Goodman <[email protected]>

manager/src/grype_db_manager/cli/db.py

@@ -8,7 +8,7 @@
 from yardstick.cli import config as ycfg
 from yardstick.cli.validate import validate as yardstick_validate
 
-from grype_db_manager import db, s3utils, grypedb
+from grype_db_manager import db, grypedb, s3utils
 from grype_db_manager.cli import config, error
 from grype_db_manager.db.format import Format
 from grype_db_manager.grypedb import DB_DIR, DBManager, GrypeDB
@@ -60,6 +60,7 @@ def remove_db(cfg: config.Application, db_uuid: str) -> None:
         click.echo(f"database {db_uuid!r} removed")
     click.echo(f"no database found with session id {db_uuid}")
 
+
 @group.command(name="build", help="build and validate a grype database")
 @click.option("--schema-version", "-s", required=True, help="the DB schema version to build")
 @click.pass_obj
@@ -131,7 +132,8 @@ def validate_db(
             db_manager.validate_namespaces(db_uuid=db_uuid)
         else:
             # TODO: implement me
-            raise NotImplementedError("namespace validation for schema v6+ is not yet implemented")
+            msg = "namespace validation for schema v6+ is not yet implemented"
+            raise NotImplementedError(msg)
 
     _validate_db(ctx, cfg, db_info, images, db_uuid, verbosity, recapture)
 
@@ -143,10 +145,19 @@ def validate_db(
     click.echo(f"{Format.BOLD}{Format.OKGREEN}Validation passed{Format.RESET}")
 
 
-def _validate_db(ctx: click.Context, cfg: config.Application, db_info: grypedb.DBInfo, images: list[str], db_uuid: str, verbosity: int, recapture: bool) -> None:
+def _validate_db(
+    ctx: click.Context,
+    cfg: config.Application,
+    db_info: grypedb.DBInfo,
+    images: list[str],
+    db_uuid: str,
+    verbosity: int,
+    recapture: bool,
+) -> None:
     if db_info.schema_version >= 6:
         # TODO: not implemented yet
-        raise NotImplementedError("validation for schema v6+ is not yet implemented")
+        msg = "validation for schema v6+ is not yet implemented"
+        raise NotImplementedError(msg)
 
     # resolve tool versions and install them
     yardstick.store.config.set_values(store_root=cfg.data.yardstick_root)
@@ -260,7 +271,8 @@ def upload_db(cfg: config.Application, db_uuid: str, ttl_seconds: int) -> None:
 
     if db_info.schema_version >= 6:
         if not os.path.exists(db_info.archive_path):
-            raise ValueError(f"latest.json file not found for DB {db_uuid!r}")
+            msg = f"latest.json file not found for DB {db_uuid!r}"
+            raise ValueError(msg)
 
         # /databases -> /databases/v6 , and is dynamic based on the schema version
         s3_path = f"{s3_path}/v{db_info.schema_version}"
@@ -282,7 +294,7 @@ def upload_db(cfg: config.Application, db_uuid: str, ttl_seconds: int) -> None:
             bucket=s3_bucket,
             key=latest_key,
             path=db_info.latest_path,
-            CacheControl=f"public,max-age=300", # 5 minutes
+            CacheControl="public,max-age=300",  # 5 minutes
         )
 
         click.echo(f"DB latest.json {db_uuid!r} uploaded to s3://{s3_bucket}/{s3_path}")

manager/src/grype_db_manager/db/latest.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import contextlib
-import datetime
 import functools
 import json
 import logging
@@ -17,10 +16,12 @@
 from grype_db_manager import grype
 
 if TYPE_CHECKING:
+    import datetime
     from collections.abc import Iterator
 
 LATEST_FILENAME = "latest.json"
 
+
 # Latest is a dataclass that represents the latest.json document for schemas v6.
 @dataclass
 class Latest:
@@ -39,7 +40,6 @@ class Latest:
     # self-describing digest of the database archive referenced in path
     checksum: str = ""
 
-
     @classmethod
     def from_json(cls, contents: str) -> Latest:
         return cls.from_dict(json.loads(contents))
@@ -82,24 +82,25 @@ def serve() -> None:
         pass
 
 
-def _log_dir(path: str, prefix: str = ""):
+def _log_dir(path: str, prefix: str = "") -> None:
     items = sorted(os.listdir(path))
     for i, item in enumerate(items):
-        is_last = (i == len(items) - 1)
+        is_last = i == len(items) - 1
         connector = "└── " if is_last else "├── "
         logging.info(f"{prefix}{connector}{item}")
         new_prefix = prefix + ("    " if is_last else "│   ")
         item_path = os.path.join(path, item)
         if os.path.isdir(item_path):
             _log_dir(item_path, new_prefix)
 
+
 def _smoke_test(
-        schema_version: str,
-        listing_url: str,
-        image: str,
-        minimum_packages: int,
-        minimum_vulnerabilities: int,
-        store_root: str,
+    schema_version: str,
+    listing_url: str,
+    image: str,
+    minimum_packages: int,
+    minimum_vulnerabilities: int,
+    store_root: str,
 ) -> None:
     logging.info(f"testing grype schema-version={schema_version!r}")
     tool_obj = grype.Grype(
@@ -125,11 +126,11 @@ def _smoke_test(
 
 
 def smoke_test(
-        test_latest: Latest,
-        archive_path: str,
-        image: str,
-        minimum_packages: int,
-        minimum_vulnerabilities: int,
+    test_latest: Latest,
+    archive_path: str,
+    image: str,
+    minimum_packages: int,
+    minimum_vulnerabilities: int,
 ) -> None:
     # write the listing to a temp dir that is served up locally on an HTTP server. This is used by grype to locally
     # download the latest.json file and check that it works against S3 (since the listing entries have DB urls that
@@ -141,7 +142,7 @@ def smoke_test(
 
         major_version = test_latest.schema_version.split(".")[0]
 
-        sub_path = os.path.join(tempdir, "v"+major_version)
+        sub_path = os.path.join(tempdir, "v" + major_version)
         os.makedirs(sub_path, exist_ok=True)
 
         logging.info(listing_contents)
@@ -152,7 +153,6 @@ def smoke_test(
         archive_dest = os.path.join(sub_path, test_latest.path)
         os.link(archive_path, archive_dest)
 
-
         # ensure grype can perform a db update for all supported schema versions. Note: we are only testing the
         # latest.json for the DB is usable (the download succeeds and grype and the update process, which does
         # checksum verifications, passes). This test does NOT check the integrity of the DB since that has already
@@ -166,4 +166,3 @@ def smoke_test(
                 minimum_vulnerabilities=minimum_vulnerabilities,
                 store_root=installation_path,
             )
-

manager/src/grype_db_manager/db/listing.py

@@ -24,6 +24,7 @@
 
 LISTING_FILENAME = "listing.json"
 
+
 # Entry is a dataclass that represents a single entry from a listing.json for schemas v1-v5.
 @dataclass
 class Entry:
@@ -45,6 +46,7 @@ def age_in_days(self, now: datetime.datetime | None = None) -> int:
             now = datetime.datetime.now(tz=datetime.timezone.utc)
         return (now - iso8601.parse_date(self.built)).days
 
+
 # Listing is a dataclass that represents the listing.json for schemas v1-v5.
 @dataclass
 class Listing:

manager/src/grype_db_manager/db/metadata.py

@@ -9,6 +9,7 @@
 
 FILE = "metadata.json"
 
+
 # Metadata is a dataclass that represents the metadata.json for schemas v1-v5.
 @dataclass
 class Metadata:

manager/src/grype_db_manager/grype.py

@@ -49,9 +49,11 @@ def _env(self, env: dict[str, str] | None = None) -> dict[str, str]:
         if not env:
             env = os.environ.copy()
         if self.schema_version >= 6:
-            env.update({
-                "GRYPE_EXP_DBV6": "true",
-            })
+            env.update(
+                {
+                    "GRYPE_EXP_DBV6": "true",
+                },
+            )
         return env
 
     def update_db(self) -> None:

manager/src/grype_db_manager/grypedb.py

@@ -385,6 +385,7 @@ def remove_db(self, db_uuid: str) -> bool:
             return True
         return False
 
+
 def db_metadata(build_dir: str) -> dict:
     metadata_path = os.path.join(build_dir, "metadata.json")
 
@@ -414,18 +415,19 @@ def db_metadata(build_dir: str) -> dict:
             # }
             return {
                 "version": int(metadata["schemaVersion"].split(".")[0]),
-                "db_checksum": None, # we don't have this information
+                "db_checksum": None,  # we don't have this information
                 "db_created": metadata["built"],
                 "data_created": parse_datetime(metadata["path"].split("_")[2]),
                 "latest_path": os.path.abspath(latest_path),
             }
 
-    msg = f"missing metadata.json and latest.json for DB"
+    msg = "missing metadata.json and latest.json for DB"
     raise DBInvalidException(msg)
 
 
 def parse_datetime(s: str) -> datetime.datetime:
-    return datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ")
+    return datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.timezone.utc)
+
 
 class GrypeDB:
     def __init__(self, bin_path: str, config_path: str = ""):

manager/tests/cli/conftest.py

@@ -8,6 +8,7 @@
 from contextlib import contextmanager
 from tempfile import TemporaryDirectory
 
+
 class Format(Enum):
     RESET = "\033[0m"
     GREEN = "\033[1;32m"
@@ -21,6 +22,7 @@ class Format(Enum):
     def render(self, text: str) -> str:
         return f"{self.value}{text}{Format.RESET.value}"
 
+
 class CustomLogger(logging.Logger):
 
     def __init__(self, name, level=logging.NOTSET):
@@ -32,6 +34,7 @@ def step(self, message: str):
             message = f"[{self.test_function}] {message}"
         self.info(Format.GREEN.render(message))
 
+
 @pytest.fixture(scope="function")
 def logger(request):
     logging.setLoggerClass(CustomLogger)
@@ -43,6 +46,7 @@ def logger(request):
 
     return logger
 
+
 @pytest.fixture(scope="function", autouse=True)
 def change_to_cli_dir(request):
     """
@@ -69,7 +73,6 @@ def change_to_cli_dir(request):
         os.chdir(original_dir)  # revert to the original directory
 
 
-
 @pytest.fixture(scope="session")
 def temporary_dir() -> str:
     with TemporaryDirectory() as tmp_dir:
@@ -82,6 +85,7 @@ def cli_env() -> dict[str, str]:
     env["PATH"] = f"{os.path.abspath('bin')}:{env['PATH']}"  # add `bin` to PATH
     return env
 
+
 class CommandHelper:
 
     def __init__(self, logger: logging.Logger):
@@ -135,10 +139,12 @@ def log_lines(text: str, prefix: str, lgr, renderer=None):
             msg = renderer(msg)
         lgr(msg)
 
+
 @pytest.fixture
 def command(logger) -> CommandHelper:
     return CommandHelper(logger)
 
+
 class GrypeHelper:
     def __init__(self, bin_dir: str | Path | None = None):
         if bin_dir:
@@ -193,6 +199,7 @@ def install(self, branch_or_version: str, bin_dir: str | None = None, env: dict[
 
         return GrypeHelper(bin_dir)
 
+
 @pytest.fixture(scope="session")
 def grype():
     return GrypeHelper()

manager/tests/cli/test_legacy_workflows.py

@@ -1,5 +1,6 @@
 import pytest
 
+
 @pytest.mark.usefixtures("cli_env")
 def test_workflow_1(cli_env, command, logger):
     """
@@ -96,12 +97,14 @@ def test_workflow_3(cli_env, command, logger, tmp_path, grype):
     bin_dir = tmp_path / "bin"
     bin_dir.mkdir(parents=True, exist_ok=True)
 
-    cli_env.update({
-        "AWS_ACCESS_KEY_ID": "test",
-        "AWS_SECRET_ACCESS_KEY": "test",
-        "AWS_REGION": "us-west-2",
-        "PATH": f"{bin_dir}:{cli_env['PATH']}",  # ensure `bin` directory is in PATH
-    })
+    cli_env.update(
+        {
+            "AWS_ACCESS_KEY_ID": "test",
+            "AWS_SECRET_ACCESS_KEY": "test",
+            "AWS_REGION": "us-west-2",
+            "PATH": f"{bin_dir}:{cli_env['PATH']}",  # ensure `bin` directory is in PATH
+        }
+    )
 
     grype = grype.install("v0.65.0", bin_dir)
 
@@ -119,10 +122,12 @@ def test_workflow_3(cli_env, command, logger, tmp_path, grype):
     assert "listing.json uploaded to s3://testbucket/grype/databases" in stdout
 
     # setup grype for DB updates and scans
-    cli_env.update({
-        "GRYPE_DB_UPDATE_URL": "http://localhost:4566/testbucket/grype/databases/listing.json",
-        "GRYPE_DB_CACHE_DIR": str(bin_dir)
-    })
+    cli_env.update(
+        {
+            "GRYPE_DB_UPDATE_URL": "http://localhost:4566/testbucket/grype/databases/listing.json",
+            "GRYPE_DB_CACHE_DIR": str(bin_dir),
+        }
+    )
 
     # validate grype DB listing and scanning
     stdout, _ = grype.run(f"db list", env=cli_env)
@@ -154,15 +159,17 @@ def test_workflow_4(cli_env, command, logger, tmp_path, grype):
     bin_dir = tmp_path / "bin"
     bin_dir.mkdir(parents=True, exist_ok=True)
 
-    cli_env.update({
-        "AWS_ACCESS_KEY_ID": "test",
-        "AWS_SECRET_ACCESS_KEY": "test",
-        "AWS_REGION": "us-west-2",
-        "SCHEMA_VERSION": "5",
-        "GRYPE_DB_MANAGER_VALIDATE_LISTING_OVERRIDE_GRYPE_VERSION": "v0.65.0",
-        "GRYPE_DB_MANAGER_VALIDATE_LISTING_OVERRIDE_DB_SCHEMA_VERSION": "5",
-        "PATH": f"{bin_dir}:{cli_env['PATH']}",  # ensure `bin` directory is in PATH
-    })
+    cli_env.update(
+        {
+            "AWS_ACCESS_KEY_ID": "test",
+            "AWS_SECRET_ACCESS_KEY": "test",
+            "AWS_REGION": "us-west-2",
+            "SCHEMA_VERSION": "5",
+            "GRYPE_DB_MANAGER_VALIDATE_LISTING_OVERRIDE_GRYPE_VERSION": "v0.65.0",
+            "GRYPE_DB_MANAGER_VALIDATE_LISTING_OVERRIDE_DB_SCHEMA_VERSION": "5",
+            "PATH": f"{bin_dir}:{cli_env['PATH']}",  # ensure `bin` directory is in PATH
+        }
+    )
 
     grype = grype.install("v0.65.0", bin_dir)
 
@@ -195,10 +202,12 @@ def test_workflow_4(cli_env, command, logger, tmp_path, grype):
     assert "listing.json uploaded to s3://testbucket/grype/databases" in stdout
 
     # set grype environment variables
-    cli_env.update({
-        "GRYPE_DB_UPDATE_URL": "http://localhost:4566/testbucket/grype/databases/listing.json",
-        "GRYPE_DB_CACHE_DIR": str(bin_dir),
-    })
+    cli_env.update(
+        {
+            "GRYPE_DB_UPDATE_URL": "http://localhost:4566/testbucket/grype/databases/listing.json",
+            "GRYPE_DB_CACHE_DIR": str(bin_dir),
+        }
+    )
 
     # validate grype DB listing and scanning
     stdout, _ = grype.run("db list", env=cli_env)