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
136 changes: 136 additions & 0 deletions tests/test_package_integration.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
import subprocess
import sys
from importlib.metadata import PackageNotFoundError

import pytest

from codealmanac.integrations.updates import package as package_module
from codealmanac.integrations.updates.package import (
InstalledPackageMetadataProvider,
SubprocessPackageCommandRunner,
clean_optional_text,
read_direct_url,
)


class FakeDistribution:
def __init__(self, version: str, files: dict[str, str]):
self.version = version
self.files = files

def read_text(self, filename: str) -> str | None:
return self.files.get(filename)


def install_distribution(monkeypatch, distribution: FakeDistribution | None) -> None:
def read(name: str) -> FakeDistribution:
if distribution is None:
raise PackageNotFoundError(name)
return distribution

monkeypatch.setattr(package_module, "distribution", read)


def test_metadata_provider_reports_unknown_when_package_is_missing(monkeypatch):
install_distribution(monkeypatch, None)

metadata = InstalledPackageMetadataProvider().read()

assert metadata.version == "unknown"
assert metadata.installer is None
assert metadata.editable is False
assert metadata.source_url is None


def test_metadata_provider_reads_installer_and_editable_direct_url(monkeypatch):
install_distribution(
monkeypatch,
FakeDistribution(
version="0.4.7",
files={
"INSTALLER": " uv \n",
"direct_url.json": (
'{"url": "file:///repo", "dir_info": {"editable": true}}'
),
},
),
)

metadata = InstalledPackageMetadataProvider().read()

assert metadata.version == "0.4.7"
assert metadata.installer == "uv"
assert metadata.editable is True
assert metadata.source_url == "file:///repo"


def test_metadata_provider_ignores_blank_installer_and_absent_direct_url(monkeypatch):
install_distribution(
monkeypatch,
FakeDistribution(version="0.4.7", files={"INSTALLER": " "}),
)

metadata = InstalledPackageMetadataProvider().read()

assert metadata.installer is None
assert metadata.editable is False
assert metadata.source_url is None


@pytest.mark.parametrize(
"raw",
[
None,
"not json",
'["pip"]',
'{"dir_info": "editable"}',
'{"url": " "}',
'{"url": 7}',
],
)
def test_read_direct_url_falls_back_to_defaults_for_unusable_payloads(raw):
metadata = read_direct_url(raw)

assert metadata.editable is False
assert metadata.source_url is None


def test_read_direct_url_requires_explicit_editable_flag():
metadata = read_direct_url('{"url": "file:///repo", "dir_info": {"editable": 1}}')

assert metadata.editable is False
assert metadata.source_url == "file:///repo"


@pytest.mark.parametrize(
("value", "expected"),
[(None, None), ("", None), (" ", None), (" pip ", "pip")],
)
def test_clean_optional_text(value, expected):
assert clean_optional_text(value) == expected


def test_command_runner_captures_output_streams_and_exit_code():
result = SubprocessPackageCommandRunner().run(
(
sys.executable,
"-c",
"import sys; print('out'); print('err', file=sys.stderr); sys.exit(3)",
)
)

assert result.exit_code == 3
assert result.stdout.strip() == "out"
assert result.stderr.strip() == "err"


def test_command_runner_reports_os_error_as_failed_result(monkeypatch):
def explode(*args, **kwargs):
raise FileNotFoundError("no such tool")

monkeypatch.setattr(subprocess, "run", explode)

result = SubprocessPackageCommandRunner().run(("missing-tool", "install"))

assert result.exit_code == 1
assert result.stderr == "FileNotFoundError: no such tool"
116 changes: 116 additions & 0 deletions tests/test_render_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
import json
from datetime import timedelta
from pathlib import Path

from codealmanac.cli.render.config import (
render_config_apply,
render_config_entry,
render_config_set,
render_config_values,
)
from codealmanac.services.automation.models import (
AutomationTask,
AutomationTaskApplyResult,
)
from codealmanac.services.config.models import (
ConfigApplyResult,
ConfigEntry,
ConfigKey,
ConfigSetResult,
)


def apply_result(task: AutomationTask, enabled: bool) -> AutomationTaskApplyResult:
return AutomationTaskApplyResult(
task=task,
enabled=enabled,
interval=timedelta(hours=1),
plist_path=Path(f"/tmp/{task.value}.plist"),
changed=True,
)


def test_render_config_values_prints_a_key_value_table(capsys):
render_config_values(
(
ConfigEntry(key=ConfigKey.AUTO_COMMIT, value="true"),
ConfigEntry(key=ConfigKey.HARNESS_DEFAULT, value="claude"),
),
json_output=False,
)

lines = capsys.readouterr().out.splitlines()
assert lines[0].split() == ["KEY", "VALUE"]
assert lines[1].split() == ["auto_commit", "true"]
assert lines[2].split() == ["harness.default", "claude"]


def test_render_config_values_json_emits_rows(capsys):
render_config_values(
(ConfigEntry(key=ConfigKey.AUTO_COMMIT, value="true"),),
json_output=True,
)

assert json.loads(capsys.readouterr().out) == [
{"key": "auto_commit", "value": "true"}
]


def test_render_config_entry_prints_only_the_value(capsys):
render_config_entry(
ConfigEntry(key=ConfigKey.HARNESS_MODEL, value="sonnet"),
json_output=False,
)

assert capsys.readouterr().out == "sonnet\n"


def test_render_config_entry_json_emits_the_entry(capsys):
render_config_entry(
ConfigEntry(key=ConfigKey.HARNESS_MODEL, value="sonnet"),
json_output=True,
)

assert json.loads(capsys.readouterr().out)["key"] == "harness.model"


def test_render_config_set_echoes_the_assignment(capsys):
result = ConfigSetResult(
path="/tmp/config.toml",
key=ConfigKey.AUTO_COMMIT,
value="false",
)

render_config_set(result, json_output=False)
render_config_set(result, json_output=True)

out = capsys.readouterr().out.splitlines()
assert out[0] == "config: auto_commit = false"
assert json.loads("\n".join(out[1:]))["value"] == "false"


def test_render_config_apply_lists_automation_task_states(capsys):
result = ConfigApplyResult(
path="/tmp/config.toml",
automation=(
apply_result(AutomationTask.SYNC, True),
apply_result(AutomationTask.GARDEN, False),
),
)

render_config_apply(result, json_output=False)

assert capsys.readouterr().out.splitlines() == [
"config applied",
" sync: enabled",
" garden: disabled",
]


def test_render_config_apply_json_emits_the_result(capsys):
render_config_apply(
ConfigApplyResult(path="/tmp/config.toml", automation=()),
json_output=True,
)

assert json.loads(capsys.readouterr().out)["path"] == "/tmp/config.toml"
Loading
Loading