Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 44 additions & 9 deletions db-auto-backup.py
Original file line number Diff line number Diff line change
Expand Up @@ -175,17 +175,52 @@ def get_backup_provider(container_names: Iterable[str]) -> Optional[BackupProvid
return None


def get_container_names(container: Container) -> Iterable[str]:
names = set()
for tag in container.image.tags:
registry, image = docker.auth.resolve_repository_name(tag)
def _name_from_image_reference(reference: str) -> Optional[str]:
"""
Extract the image name from a tag or digest reference.

Handles three forms:
- ``name:tag`` (e.g. ``postgres:14-alpine``)
- ``name@digest`` (e.g. ``ghcr.io/immich-app/postgres@sha256:abc...``)
- ``registry/name:tag@digest`` (e.g. ``docker.io/library/postgres:14@sha256:abc...``)
"""
# Strip the digest suffix if present (e.g. @sha256:...).
# We only care about the name part for matching against patterns.
image_ref = reference.split("@", 1)[0]

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestion: Split from the right

Suggested change
image_ref = reference.split("@", 1)[0]
image_ref = reference.rsplit("@", 1)[0]

if not image_ref:
return None

registry, image = docker.auth.resolve_repository_name(image_ref)

# HACK: Strip "library" from official images
if registry == docker.auth.INDEX_NAME:
image = image.removeprefix("library/")
# HACK: Strip "library" from official images
if registry == docker.auth.INDEX_NAME:
image = image.removeprefix("library/")

# Tags may be absent when the image was referenced only by digest, so
# `:` may not appear. In that case, the whole string is the name.
if ":" in image:
image, _ = image.split(":", 1)

return image


def get_container_names(container: Container) -> Iterable[str]:
names: set[str] = set()

# Prefer `RepoTags` for the common case. When the image was pulled by
# digest, this is empty and we fall back to the original image reference
# stored on the container's config, which preserves the digest form.
references: list[str] = list(container.image.tags)
if not references:
config_image = (container.attrs.get("Config") or {}).get("Image")
if config_image:
references.append(config_image)

for reference in references:
name = _name_from_image_reference(reference)
if name:
names.add(name)

image, tag_name = image.split(":", 1)
names.add(image)
return names


Expand Down
69 changes: 69 additions & 0 deletions tests/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -135,3 +135,72 @@ def test_get_backup_provider(container_name: str, name: str) -> None:

assert provider is not None
assert provider.name == name


@pytest.mark.parametrize(
"reference,name",
[
(
"ghcr.io/immich-app/postgres:14-vectorchord0.4.3-pgvectors0.2.0@sha256:"
"32324a2f41df5de9efe1af166b7008c3f55646f8d0e00d9550c16c9822366b4a",
"immich-app/postgres",
),
(
"postgres@sha256:32324a2f41df5de9efe1af166b7008c3f55646f8d0e00d9550c16c9822366b4a",
"postgres",
),
(
"docker.io/library/postgres@sha256:32324a2f41df5de9efe1af166b7008c3f55646f8d0e00d9550c16c9822366b4a",
"postgres",
),
(
"ghcr.io/realorangeone/db-auto-backup@sha256:32324a2f41df5de9efe1af166b7008c3f55646f8d0e00d9550c16c9822366b4a",
"realorangeone/db-auto-backup",
),
],
)
def test_name_from_image_reference_with_digest(reference: str, name: str) -> None:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestion: This test can be folded into test_get_container_names - I don't think the internal helper needs its own tests

assert db_auto_backup._name_from_image_reference(reference) == name


@pytest.mark.parametrize(
"reference,name",
[
("postgres:14-alpine", "postgres"),
("ghcr.io/realorangeone/db-auto-backup:latest", "realorangeone/db-auto-backup"),
("postgres", "postgres"),
("ghcr.io/immich-app/postgres", "immich-app/postgres"),
],
)
def test_name_from_image_reference_without_digest(reference: str, name: str) -> None:
assert db_auto_backup._name_from_image_reference(reference) == name


def test_get_container_names_falls_back_to_config_image() -> None:
"""
Images pulled by digest (e.g. `image@sha256:...`) have an empty
`RepoTags` list. `get_container_names` should fall back to the
container's `Config.Image` so the container is still detected.
"""
container = MagicMock()
container.image.tags = []
container.attrs = {
"Config": {
"Image": (
"ghcr.io/immich-app/postgres:14-vectorchord0.4.3-pgvectors0.2.0"
"@sha256:32324a2f41df5de9efe1af166b7008c3f55646f8d0e00d9550c16c9822366b4a"
)
}
}
assert db_auto_backup.get_container_names(container) == {"immich-app/postgres"}


def test_get_container_names_empty_when_no_reference() -> None:
"""
When both `RepoTags` and `Config.Image` are missing or empty, the
container should simply produce no names (not crash).
"""
container = MagicMock()
container.image.tags = []
container.attrs = {"Config": {}}
assert db_auto_backup.get_container_names(container) == set()
Loading