Skip to content
Open
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
28 changes: 25 additions & 3 deletions src/bluefinctl/core/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,22 @@ def _detect_gpu() -> GpuInfo:
return GpuInfo()


def _tag_from_image_ref(ref: str) -> str | None:
"""Extract the tag from an OCI image reference if present.

Returns ``None`` for digest references or bare refs without a tag.
"""
if not ref or "@" in ref:
return None
if ":" not in ref:
return None
tag = ref.rsplit(":", 1)[-1]
# Tags cannot contain '/' and cannot look like a registry port.
if "/" in tag or tag.isdigit():
return None
return tag or None


def _check_devmode() -> bool:
"""Check if developer mode is active."""
devmode_flag = Path("/etc/ublue-os/devmode")
Expand All @@ -184,9 +200,10 @@ async def get_system_info() -> SystemInfo:
raw_ref = image_data.get("image-ref", "")
image_signed = raw_ref.startswith(_SIGNED_PREFIX)

# Bootc status — staged update + hostname
# Bootc status — runtime image ref, staged update + hostname
boot_status = "Current"
image_staged = False
runtime_tag: str | None = None
hostname = ""
try:
proc = await asyncio.create_subprocess_exec(
Expand All @@ -197,10 +214,15 @@ async def get_system_info() -> SystemInfo:
stdout, _ = await proc.communicate()
if proc.returncode == 0:
bootc_data = json.loads(stdout)
staged = bootc_data.get("status", {}).get("staged")
status = bootc_data.get("status", {})
staged = status.get("staged")
if staged:
image_staged = True
boot_status = "Update staged — reboot to apply"
booted_image = status.get("booted", {}).get("image", {}).get("image", {})
runtime_ref = booted_image.get("image", "")
if runtime_ref:
runtime_tag = _tag_from_image_ref(runtime_ref)
except (FileNotFoundError, OSError):
boot_status = "bootc unavailable"

Expand All @@ -209,7 +231,7 @@ async def get_system_info() -> SystemInfo:

return SystemInfo(
image_name=image_data.get("image-name", "unknown"),
image_tag=image_data.get("image-tag", "unknown"),
image_tag=runtime_tag or image_data.get("image-tag", "unknown"),
image_ref=raw_ref,
boot_status=boot_status,
image_staged=image_staged,
Expand Down
123 changes: 123 additions & 0 deletions tests/test_system.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,123 @@
"""Tests for bluefinctl.core.system image-tag resolution.

Covers:
- _tag_from_image_ref helper parses OCI refs correctly
- get_system_info prefers the runtime bootc tag over build-time image-info.json
- get_system_info falls back to image-info.json when bootc status lacks a tag
"""

from __future__ import annotations

import json
from typing import Any
from unittest.mock import AsyncMock, patch

import pytest

from bluefinctl.core.system import SystemInfo, _tag_from_image_ref, get_system_info


def _make_image_info(tag: str = "latest") -> dict[str, Any]:
return {
"image-name": "bluefin",
"image-tag": tag,
"image-ref": "ostree-image-signed:docker://ghcr.io/projectbluefin/bluefin",
}


def _bootc_status_json(image: str) -> bytes:
return json.dumps(
{
"status": {
"booted": {
"image": {
"image": {"image": image, "digest": "sha256:c0ffee"},
},
},
},
}
).encode()


class TestTagFromImageRef:
def test_extracts_standard_tag(self) -> None:
assert _tag_from_image_ref("ghcr.io/projectbluefin/bluefin:testing") == "testing"

def test_extracts_registry_port_tag(self) -> None:
assert _tag_from_image_ref("registry:5000/bluefin:testing") == "testing"

def test_returns_none_for_digest_ref(self) -> None:
assert _tag_from_image_ref("ghcr.io/projectbluefin/bluefin@sha256:c0ffee") is None

def test_returns_none_for_bare_ref(self) -> None:
assert _tag_from_image_ref("ghcr.io/projectbluefin/bluefin") is None


class TestGetSystemInfoImageTag:
@pytest.mark.asyncio
async def test_prefers_bootc_status_tag(self) -> None:
"""Runtime tag from bootc status should override build-time image-info."""
proc = AsyncMock()
proc.returncode = 0
proc.communicate = AsyncMock(
return_value=(_bootc_status_json("ghcr.io/projectbluefin/bluefin:testing"), b"")
)

with (
patch(
"bluefinctl.core.system._read_image_info", return_value=_make_image_info("latest")
),
patch("bluefinctl.core.system._detect_gpu", return_value=AsyncMock()),
patch("bluefinctl.core.system._check_devmode", return_value=False),
patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)),
patch("socket.gethostname", return_value="test-host"),
):
info = await get_system_info()

assert isinstance(info, SystemInfo)
assert info.image_tag == "testing"

@pytest.mark.asyncio
async def test_falls_back_to_image_info_when_no_bootc_tag(self) -> None:
"""When bootc status has no tag, keep build-time image-info tag."""
proc = AsyncMock()
proc.returncode = 0
proc.communicate = AsyncMock(
return_value=(_bootc_status_json("ghcr.io/projectbluefin/bluefin"), b"")
)

with (
patch(
"bluefinctl.core.system._read_image_info", return_value=_make_image_info("latest")
),
patch("bluefinctl.core.system._detect_gpu", return_value=AsyncMock()),
patch("bluefinctl.core.system._check_devmode", return_value=False),
patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)),
patch("socket.gethostname", return_value="test-host"),
):
info = await get_system_info()

assert isinstance(info, SystemInfo)
assert info.image_tag == "latest"

@pytest.mark.asyncio
async def test_falls_back_to_image_info_when_bootc_missing(self) -> None:
"""If bootc CLI is unavailable, use build-time image-info tag."""
proc = AsyncMock()
proc.returncode = 1
proc.communicate = AsyncMock(return_value=(b"", b"no bootc"))

with (
patch(
"bluefinctl.core.system._read_image_info",
return_value={"image-name": "bluefin", "image-tag": "stable"},
),
patch("bluefinctl.core.system._detect_gpu", return_value=AsyncMock()),
patch("bluefinctl.core.system._check_devmode", return_value=False),
patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)),
patch("socket.gethostname", return_value="test-host"),
):
info = await get_system_info()

assert isinstance(info, SystemInfo)
assert info.image_tag == "stable"