write some tests

.github/workflows/pytests.yaml

@@ -0,0 +1,26 @@
+name: pytests
+
+on: [push, pull_request]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    name: Pytest
+    if: github.event_name == 'pull_request' || github.ref == 'refs/heads/main'
+    defaults:
+      run:
+        working-directory: .
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        submodules: recursive
+    - uses: actions/setup-python@v5
+      with:
+        python-version: '3.12'
+        cache: 'pip'
+    - name: Install requirements
+      run: python3 -m pip install -r requirements.txt -r requirements-dev.txt
+    - name: Run tests and collect coverage
+      run: python3 -m pytest --cov --cov-report xml .
+    - name: Upload coverage reports to Codecov with GitHub Action
+      uses: codecov/codecov-action@v4

containercrop/main.py

@@ -3,12 +3,14 @@
 and keep only the images you need.
 """
 
+import logging
 import os
 from datetime import datetime
+from fnmatch import fnmatch
 from typing import Annotated
 
 from dateparser import parse
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 from .github_api import Image
 
@@ -34,7 +36,7 @@ class RetentionArgs(BaseModel):
     token: str | None = None
     untagged_only: bool = False
     skip_tags: list[str]
-    keep_at_least: Annotated[int, Field(ge=0)]
+    keep_at_least: Annotated[int, Field(ge=0)] = 0
     filter_tags: list[str] = Field(default_factory=list)
     dry_run: bool = False
     repo_owner: str
@@ -68,15 +70,43 @@ def parse_human_readable_datetime(cls, v: str) -> datetime:
             raise ValueError("Timezone is required for the cut-off")
         return parsed_cutoff
 
+    @model_validator(mode="after")
+    def check_skip_tags_and_untagged_only(self) -> "RetentionArgs":
+        if self.untagged_only and self.skip_tags:
+            raise ValueError("Cannot set both `untagged_only` and `skip_tags`.")
+        return self
+
+
+def matches_retention_policy(image: Image, args: RetentionArgs) -> bool:
+    """
+    Check if the image matches the retention policy.
+    :param image: The image to check
+    :param args: The retention policy
+    :return: True if the image should be deleted
+    """
+    if args.skip_tags and any(
+        any(fnmatch(tag, skip_tag) for skip_tag in args.skip_tags) for tag in image.tags
+    ):
+        return False
+    if args.untagged_only and image.tags:
+        return False
+    if args.cut_off and image.is_before_cut_off_date(args.cut_off):
+        return True
+    if args.filter_tags and any(
+        any(fnmatch(tag, filter_tag) for filter_tag in args.filter_tags)
+        for tag in image.tags
+    ):
+        logging.debug(f"Image {image.name} does match filter tags")
+        return True
+    return False
+
 
 def apply_retention_policy(args: RetentionArgs, images: list[Image]) -> list[Image]:
     """
     Apply the retention policy to the images and return the ones that should be deleted.
     """
-    # to implement here: cut-off, untagged_only, skip_tags, keep_at_leat, filter_tags
-    # cut-off
-
-    return images[: args.keep_at_least]
+    matches = [image for image in images if matches_retention_policy(image, args)]
+    return matches[args.keep_at_least :]
 
 
 async def main():

containercrop/test_main.py

@@ -1,10 +1,33 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
+
+import pytest
 
 from .github_api import Image
 from .main import RetentionArgs, apply_retention_policy
 
 
-def test_pydantic_parsing_happy_path():
+@pytest.fixture
+def generate_images():
+    """Generate images with different properties."""
+
+    def _generator(tags_list, days_old_list, names_list):
+        return [
+            Image(
+                id=i,
+                name=names_list[i % len(names_list)],
+                created_at=datetime.now(timezone.utc)
+                - timedelta(days=days_old_list[i % len(days_old_list)]),
+                updated_at=datetime.now(timezone.utc)
+                - timedelta(days=days_old_list[i % len(days_old_list)]),
+                tags=tags_list[i % len(tags_list)],
+            )
+            for i in range(len(tags_list))
+        ]
+
+    return _generator
+
+
+def test_RetentionArgs_parsing_happy_path():
     inp = {
         "image_names": "test",
         "cut_off": "1 day ago UTC",
@@ -21,6 +44,20 @@ def test_pydantic_parsing_happy_path():
     assert args.keep_at_least == 0
 
 
+def test_RetentionArgs_cant_set_both_skip_tags_and_untagged_only():
+    inp = {
+        "image_names": "test",
+        "cut_off": "1 day ago UTC",
+        "untagged_only": "true",
+        "skip_tags": "test",
+        "filter_tags": "test",
+        "dry_run": "false",
+        "repo_owner": "test",
+    }
+    with pytest.raises(ValueError):
+        RetentionArgs(**inp)
+
+
 def test_apply_retention_policy_keep_at_least():
     policy = RetentionArgs(
         image_names="test",
@@ -34,13 +71,213 @@ def test_apply_retention_policy_keep_at_least():
         Image(
             id=i,
             name="test",
-            created_at=datetime.now() - timedelta(days=30),
-            updated_at=datetime.now() - timedelta(days=30),
+            created_at=datetime.now(timezone.utc) - timedelta(days=30),
+            updated_at=datetime.now(timezone.utc) - timedelta(days=30),
         )
         for i in range(30)
     ]
 
-    assert len(apply_retention_policy(policy, images)) == 10
+    assert len(apply_retention_policy(policy, images)) == 20
+
+    images = images[:5]
+    assert len(apply_retention_policy(policy, images)) == 0
+
+
+def test_delete_untagged_images_older_than_one_day():
+    policy = RetentionArgs(
+        image_names="test",
+        cut_off="1 day ago UTC",
+        skip_tags="",
+        keep_at_least=0,
+        repo_owner="test",
+        untagged_only=True,
+    )
+    # create 30 old images and verify that we keep 10
+    images = [
+        Image(
+            id=i,
+            name="test",
+            created_at=datetime.now(timezone.utc) - timedelta(days=30),
+            updated_at=datetime.now(timezone.utc) - timedelta(days=30),
+            tags=[],
+        )
+        for i in range(30)
+    ]
+
+    assert len(apply_retention_policy(policy, images)) == 30
 
     images = images[:5]
     assert len(apply_retention_policy(policy, images)) == 5
+
+
+def test_filter_by_specific_tags_for_deletion(generate_images):
+    images = generate_images(
+        tags_list=[["v1"], ["v1.0"], ["beta"], ["latest"]],
+        days_old_list=[5, 5, 5, 5],
+        names_list=["image1"],
+    )
+    policy = RetentionArgs(
+        image_names="image1",
+        cut_off="20 days ago UTC",
+        untagged_only=False,
+        skip_tags="",
+        keep_at_least=0,
+        filter_tags="v1,v1.0",
+        dry_run=True,
+        token="dummy_token",
+        repo_owner="test",
+    )
+    assert len(apply_retention_policy(policy, images)) == 2
+
+
+def test_skip_specific_tags(generate_images):
+    images = generate_images(
+        tags_list=[["v1"], ["beta"], ["latest"]],
+        days_old_list=[1, 2, 3],
+        names_list=["image1"],
+    )
+    policy = RetentionArgs(
+        image_names="image1",
+        cut_off="1 day ago UTC",
+        untagged_only=False,
+        skip_tags="beta,latest",
+        keep_at_least=0,
+        filter_tags="",
+        dry_run=True,
+        token="dummy_token",
+        repo_owner="test",
+    )
+    assert (
+        len(apply_retention_policy(policy, images)) == 1
+    )  # Only "v1" is eligible for deletion
+
+
+def test_keep_at_least_n_images(generate_images):
+    images = generate_images(
+        tags_list=[[], ["v1"], ["latest"]],
+        days_old_list=[10, 15, 20],
+        names_list=["image1"],
+    )
+    policy = RetentionArgs(
+        image_names="image1",
+        cut_off="5 days ago UTC",
+        untagged_only=False,
+        skip_tags="",
+        keep_at_least=2,
+        filter_tags="",
+        repo_owner="test",
+    )
+    assert (
+        len(apply_retention_policy(policy, images)) == 1
+    )  # Despite all being old, 2 must be kept
+
+
+def test_delete_untagged_images_older_than_x_days(generate_images):
+    images = generate_images(
+        tags_list=[[], [], [], ["v1"], ["latest"]],
+        days_old_list=[3, 4, 5, 1, 2],
+        names_list=["image1", "image2"],
+    )
+    policy = RetentionArgs(
+        image_names="image1,image2",
+        cut_off="2 days ago UTC",
+        untagged_only=True,
+        skip_tags="",
+        keep_at_least=0,
+        filter_tags="",
+        repo_owner="test",
+    )
+    to_delete = apply_retention_policy(policy, images)
+    assert len(to_delete) == 3
+    assert all(
+        img.tags == [] for img in to_delete
+    )  # Ensure all deleted images are untagged
+
+
+def test_delete_tagged_keep_recent_untagged(generate_images):
+    images = generate_images(
+        tags_list=[["v1"], [], ["v2"], []],
+        days_old_list=[10, 1, 10, 2],
+        names_list=["image1"],
+    )
+    policy = RetentionArgs(
+        image_names="image1",
+        cut_off="5 days ago UTC",
+        untagged_only=False,
+        skip_tags="",
+        keep_at_least=0,
+        filter_tags="",
+        repo_owner="test",
+    )
+    deleted_images = apply_retention_policy(policy, images)
+    assert len(deleted_images) == 2  # Should keep the 2 recent tagged images
+    print(deleted_images)
+    assert all(
+        img.tags != [] for img in deleted_images
+    )  # Ensure the kept images are untagged
+
+
+@pytest.mark.skip
+def test_dry_run_does_not_delete_images(generate_images):
+    images = generate_images(
+        tags_list=[["v1"], ["v2"]], days_old_list=[10, 20], names_list=["image1"]
+    )
+    initial_image_count = len(images)
+    policy = RetentionArgs(
+        image_names="image1",
+        cut_off="5 days ago UTC",
+        untagged_only=False,
+        skip_tags="",
+        keep_at_least=0,
+        filter_tags="",
+        dry_run=True,  # Important: This is a dry run
+        token="dummy_token",
+        repo_owner="test",
+    )
+    images = apply_retention_policy(policy, images)
+    # Assuming apply_retention_policy modifies the list of images when not in dry run
+    assert (
+        len(images) == initial_image_count
+    )  # Ensure no images are actually deleted in dry run
+
+
+def test_keep_all_images_when_keep_at_least_equals_total(generate_images):
+    images = generate_images(
+        tags_list=[[], ["v1"]], days_old_list=[30, 30], names_list=["image1", "image2"]
+    )
+    policy = RetentionArgs(
+        image_names="image1,image2",
+        cut_off="1 day ago UTC",
+        untagged_only=False,
+        skip_tags="",
+        keep_at_least=len(images),  # Set to keep all images
+        filter_tags="",
+        repo_owner="test",
+    )
+    assert (
+        len(apply_retention_policy(policy, images)) == 0
+    )  # Expect no images to be marked for deletion
+
+
+def test_wildcard_tag_filtering(generate_images):
+    images = generate_images(
+        tags_list=[["v1.1"], ["v1.2"], ["beta"], ["latest"]],
+        days_old_list=[10, 15, 5, 20],
+        names_list=["image1"],
+    )
+    policy = RetentionArgs(
+        image_names="image1",
+        cut_off="3 days ago UTC",
+        untagged_only=False,
+        skip_tags="v1.*",  # Skip any version starting with v1.
+        keep_at_least=0,
+        filter_tags="*",  # Consider all tags
+        repo_owner="test",
+    )
+    retained_images = apply_retention_policy(policy, images)
+    assert (
+        len(retained_images) == 2
+    )  # "beta" and possibly "latest" if not matching skip_tags wildcard
+    assert not any(
+        "v1" in tag for img in retained_images for tag in img.tags
+    )  # No v1.* tags

requirements-dev.txt

@@ -1,2 +1,3 @@
 pytest
-pytest-asyncio
+pytest-asyncio
+pytest-cov