diff --git a/.github/workflows/k8s-version-matrix.yml b/.github/workflows/k8s-version-matrix.yml index c12e87804..c64a3a293 100644 --- a/.github/workflows/k8s-version-matrix.yml +++ b/.github/workflows/k8s-version-matrix.yml @@ -16,6 +16,11 @@ on: - src/korvid/core/capability_classifier.py - src/korvid/core/config_consumers.py - src/korvid/core/config_propagation.py + - src/korvid/core/creation_constraints.py + - src/korvid/core/creation_failures.py + - src/korvid/core/creation_policies.py + - src/korvid/core/creation_quantities.py + - src/korvid/core/creation_workloads.py - src/korvid/core/evidence_reads.py - src/korvid/core/run_tokens.py - src/korvid/k8s/bounded_transport.py @@ -24,6 +29,7 @@ on: - src/korvid/k8s/discovery.py - src/korvid/k8s/discovery_snapshot.py - src/korvid/k8s/errors.py + - src/korvid/k8s/events.py - src/korvid/k8s/evidence_requests.py - tests/version_matrix/** - .github/workflows/k8s-version-matrix.yml @@ -37,6 +43,11 @@ on: - src/korvid/core/capability_classifier.py - src/korvid/core/config_consumers.py - src/korvid/core/config_propagation.py + - src/korvid/core/creation_constraints.py + - src/korvid/core/creation_failures.py + - src/korvid/core/creation_policies.py + - src/korvid/core/creation_quantities.py + - src/korvid/core/creation_workloads.py - src/korvid/core/evidence_reads.py - src/korvid/core/run_tokens.py - src/korvid/k8s/bounded_transport.py @@ -45,6 +56,7 @@ on: - src/korvid/k8s/discovery.py - src/korvid/k8s/discovery_snapshot.py - src/korvid/k8s/errors.py + - src/korvid/k8s/events.py - src/korvid/k8s/evidence_requests.py - tests/version_matrix/** - .github/workflows/k8s-version-matrix.yml diff --git a/docs/keybindings.md b/docs/keybindings.md index 39d8bd503..755dd768b 100644 --- a/docs/keybindings.md +++ b/docs/keybindings.md @@ -12,6 +12,7 @@ Korvid shows only the keys that act on the current view. Press `?` for the compl | `:pulse` / `:problems` | Inspect [current problems, recent warnings, and coverage](pulse.md) | | `:outcomes` / `:deployment-outcomes` | Open the latest bounded Deployment scale/restart outcome | | `:consumers` / `:config-consumers` | Show which Pods use the selected ConfigMap or Secret, and [what a change means](tui.md#configuration-propagation) for each | +| `:blocked` | Show why the selected workload's Pods are [not being created](tui.md#workload-creation-failures): recorded failures, quotas and LimitRanges | | `Ctrl-P` | Search app actions and built-in commands by intent | | `:keys` / `:keybindings` | Edit, review, and save app keybindings | | `:slots` / `:ns-slots` | Preview and repack the `1`–`9` namespace shortcuts | diff --git a/docs/kubernetes-support.md b/docs/kubernetes-support.md index 0e3fa7730..eb1a3bfa8 100644 --- a/docs/kubernetes-support.md +++ b/docs/kubernetes-support.md @@ -71,6 +71,9 @@ read-only scenarios against each: volume and through a `subPath` mount, each found in its own mode, without reading the ConfigMap's data and without the Pod's literal environment values reaching the report. +- a Deployment whose namespace's quota admits no Pod, whose ReplicaSet's + recorded rejection is read next to the quota and a LimitRange, without the + template's literal environment values reaching the report. The workflow runs on demand, and whenever a pull request or a push to `main` changes the modules those scenarios exercise or the project's dependencies. Each passing job records the diff --git a/docs/release-notes/unreleased.md b/docs/release-notes/unreleased.md index bafdf5649..719d5184c 100644 --- a/docs/release-notes/unreleased.md +++ b/docs/release-notes/unreleased.md @@ -22,6 +22,15 @@ tracks the next release scope. consumer's container start, judged with a one-minute allowance for clock skew. Only changes seen between two openings in the same session count. +## Workload creation failures + +- `:blocked` (or **Explain workload creation failures** in the Action + Palette) shows why a Deployment, ReplicaSet, StatefulSet, DaemonSet, Job or + CronJob has no Pods: the `FailedCreate` rejections recorded for it, next to + its namespace's ResourceQuotas and LimitRanges, with quantities compared + exactly and absence said only from complete lists. See + [Workload creation failures](../tui.md#workload-creation-failures). + ## Current configuration and extension contracts Existing migration links continue in the diff --git a/docs/superpowers/plans/2026-10-30-creation-constraints-view.md b/docs/superpowers/plans/2026-10-30-creation-constraints-view.md new file mode 100644 index 000000000..2ce635bca --- /dev/null +++ b/docs/superpowers/plans/2026-10-30-creation-constraints-view.md @@ -0,0 +1,1877 @@ +# Workload Creation Constraints, View (#411 Q1) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Show the selected workload's recorded creation failures next to its namespace's ResourceQuotas and LimitRanges, through `:blocked` and the Action Palette (#525, the view half of slice Q1 of #411). + +**Architecture:** `ui/widgets/creation_constraints_screen.py` renders one `CreationReport` (from #524) as pure row and note functions and a modal screen. `ui/creation_constraints_controller.py` runs `explain_creation` in a worker under a run token, as the `:consumers` controller does. A `:blocked` command and palette entry reach it through the command router; the composition root hands it the session's `EvidenceClient` and the workspace's `jump_to_object`. + +**Tech Stack:** Python 3.11+, Textual, Rich `Text`, pytest with `asyncio_mode = "auto"`, mypy `--strict`, ruff, tach. + +## Global Constraints + +- **Design:** `docs/superpowers/specs/2026-10-30-creation-constraints-design.md`, sections "The view", "Evidence" and "Controller and lifecycle"; the presentation contract of `docs/superpowers/specs/2026-10-25-capability-aware-diagnostics-design.md`. +- **One line per source** from the shared renderer, in declaration order; the children line only for a Deployment or CronJob. +- **Attribution only as far as the message goes:** a quota by the name the message gives, flagged when the list lacks it; a LimitRange as "a LimitRange in this namespace"; anything else as "other rejection". +- **Absence only from a conclusive list:** "No ResourceQuota/LimitRange in namespace …" only when that list is conclusive; otherwise "Not established" with the reason, "the list is stale" once it ages. No recorded failure is never a conclusion. +- **Literal text:** every cell, the title and the notes are plain `Text`. +- **Late results:** dropped after a context switch, new credentials or a newer run, or when the view, scope or selection changed or a dialog opened; the context switch cancels the worker group; a failed worker is reported. +- **Palette search:** the help text must not contain the subsequence `h…e…l…m`, which a test reserves for the Helm rows. +- **Size limits:** `scripts/check_source_size.py` is not edited. +- **Commands:** use `uv run --frozen …`. Behind the corporate mirror, call the project venv's tools directly with `PYTHONPATH=src`. + +## Departures from the design + +- **Four generic columns.** The view's sections differ, so the table uses `Object`, `What`, `Value` and `Note` for all three, with a heading row per section, as the `:consumers` view does. +- **Counts read "once" or "N times".** The design's sketch used `12×`; ruff rejects the ambiguous multiplication sign, and words read better. +- **`elapsed_text`.** Failure ages read "last 2m ago", so `evidence_text` gains `elapsed_text`, which `age_text` now uses. +- **Destinations carry their API group.** A row leads to a group and resource, not a view alias; the controller resolves it with `resolve_resource` and `canonical_resource_alias`, as `WorkspaceController.drill_into` does, so a custom API that claimed `replicasets` or `jobs` cannot receive the navigation. A resource the session has not discovered is said, not opened. +- **Rows grow to fit.** Data rows use automatic height, so a multiline rejection message is shown in full. +- **Unreadable policies are said.** A quota or LimitRange the projection could not read in full shows "constraints could not be read", never "no hard limits" or "no constraints". +- **The table's times refresh too.** The timer re-renders time-dependent cells (failure ages and quota attribution) with `update_cell`, keeping the cursor. +- **The fake UI surface records exclusivity.** `FakeUiSurface.exclusive_workers` lets the controller test pin `exclusive=True`, which no test checked before. + +## File structure + +| File | Change | Responsibility | +| --- | --- | --- | +| `src/korvid/ui/evidence_text.py` | modify | `elapsed_text` | +| `src/korvid/ui/widgets/creation_constraints_screen.py` | create | `Destination`, `title_text`, `evidence_lines`, `table_rows`, `report_notes`, `CreationConstraintsScreen` | +| `src/korvid/ui/creation_constraints_controller.py` | create | `CREATION_CONSTRAINTS_GROUP`, `CreationConstraintsController` | +| `src/korvid/ui/messages.py`, `command.py`, `command_router.py`, `action_policy.py` | modify | `:blocked`, its palette entry, routing and availability | +| `src/korvid/ui/context_switch_coordinator.py`, `worker_failures.py` | modify | cancel the group on a switch; report a failed run | +| `src/korvid/__main__.py` | modify | wire the controller | +| `tests/version_matrix/fixtures.yaml`, `test_scenarios.py`, `.github/workflows/k8s-version-matrix.yml` | modify | the live scenario and its path filters | +| `docs/tui.md`, `docs/keybindings.md`, `docs/kubernetes-support.md`, `docs/release-notes/unreleased.md` | modify | what the view shows and how to read it | + +--- + +### Task 1: The view + +Create `tests/ui/test_creation_constraints_screen.py` with: + +```python +"""The workload creation constraints view (#411): rows, notes and keys.""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from rich.text import Text +from textual.app import App +from textual.coordinate import Coordinate +from textual.widgets import DataTable, Static + +from korvid.core.capability import ( + ApiSource, + Baseline, + Basis, + EvidenceState, + Observation, + SourceScope, + Verb, +) +from korvid.core.creation_constraints import CreationReport, CreationTarget +from korvid.core.creation_failures import Failure, FailureSource, Subject, classify +from korvid.core.creation_policies import ( + LimitRangeProjection, + LimitValue, + QuotaEntry, + QuotaProjection, + ScopeExpression, +) +from korvid.core.creation_quantities import QuotaGroup, Usage +from korvid.core.creation_workloads import Counter, WorkloadProjection +from korvid.ui.widgets.creation_constraints_screen import ( + CreationConstraintsScreen, + Destination, + Heading, + Row, + evidence_lines, + report_notes, + table_rows, + title_text, +) +from tests.ui.waits import screen_ready, until + +NOW = datetime(2026, 10, 30, 10, 0, tzinfo=UTC) +TARGET = CreationTarget("Deployment", "shop", "api", "uid-api") +SCOPE = SourceScope.namespace("shop") +DEPLOYMENT = Subject("Deployment", "api", "uid-api") +REPLICA_SET = Subject("ReplicaSet", "api-7d9f", "uid-rs") +EXCEEDED = ( + 'Error creating: pods "api-7d9f-x" is forbidden: exceeded quota: compute, ' + "requested: pods=1, used: pods=20, limited: pods=20" +) + + +def _source(group: str, resource: str, verb: Verb = Verb.LIST, **extra: Any) -> ApiSource: + return ApiSource(group, resource, SCOPE, "v1", verb, **extra) + + +def _seen(source: ApiSource, basis: Basis = Basis.OBSERVED, uid: str | None = None) -> Observation: + state = { + Basis.OBSERVED: EvidenceState.AVAILABLE, + Basis.OBJECT_ABSENT: EvidenceState.AVAILABLE, + Basis.TRUNCATED: EvidenceState.PARTIAL, + Basis.FORBIDDEN: EvidenceState.DENIED, + Basis.STALE_STATUS: EvidenceState.PARTIAL, + }[basis] + return Observation(source, state, basis, NOW - timedelta(seconds=2), observed_uid=uid) + + +WORKLOAD_SOURCE = _source("apps", "deployments", Verb.GET, name="api", expected_uid="uid-api") +WORKLOAD = WorkloadProjection( + DEPLOYMENT, (Counter("desired", "3"), Counter("created", "0")), None, None +) +QUOTA = QuotaProjection( + "compute", + "uid-q", + ( + QuotaEntry("requests.cpu", QuotaGroup.COMPUTE, "2", "2", Usage.AT_LIMIT), + QuotaEntry("pods", QuotaGroup.COUNT, "20", "20", Usage.AT_LIMIT), + QuotaEntry("services", QuotaGroup.COUNT, "5", "1", Usage.BELOW), + ), + ("NotTerminating",), + (ScopeExpression("PriorityClass", "In", ("gold", "silver")),), +) +LIMITS = LimitRangeProjection( + "defaults", + "uid-lr", + (LimitValue("Container", "cpu", max="1", default="500m", default_request="100m"),), +) +FAILURE = Failure( + REPLICA_SET, FailureSource.EVENT, classify(EXCEEDED), 12, NOW - timedelta(minutes=2) +) + + +def _report(**changes: Any) -> CreationReport: + report = CreationReport( + target=TARGET, + workload_observation=_seen(WORKLOAD_SOURCE, uid="uid-api"), + children_observation=_seen(_source("apps", "replicasets")), + events_observation=_seen(_source("events.k8s.io", "events")), + quotas_observation=_seen(_source("", "resourcequotas")), + limit_ranges_observation=_seen(_source("", "limitranges")), + workload=WORKLOAD, + children=(), + failures=(FAILURE,), + earlier_failures=0, + quotas=(QUOTA,), + limit_ranges=(LIMITS,), + baseline=Baseline(EvidenceState.AVAILABLE), + ) + return replace(report, **changes) + + +def _plain(lines: list[Text]) -> list[str]: + return [line.plain for line in lines] + + +def _rows(report: CreationReport) -> list[Row]: + return [row for row in table_rows(report, NOW) if isinstance(row, Row)] + + +def test_the_title_names_the_workload_and_its_counters() -> None: + assert title_text(_report()) == "Deployment shop/api desired 3 · created 0" + assert title_text(_report(workload=None)) == "Deployment shop/api" + unset = replace(WORKLOAD, counters=(Counter("desired", None), Counter("created", "0"))) + assert title_text(_report(workload=unset)) == "Deployment shop/api desired ? · created 0" + + +def test_each_source_has_a_line_from_the_shared_renderer_in_order() -> None: + lines = _plain(evidence_lines(_report(), NOW)) + + assert [line.split(":")[0] for line in lines] == [ + "deployments apps/v1 in namespace shop", + "replicasets apps/v1 in namespace shop", + "events events.k8s.io/v1 in namespace shop", + "resourcequotas v1 in namespace shop", + "limitranges v1 in namespace shop", + ] + no_children = _plain(evidence_lines(_report(children_observation=None), NOW)) + assert len(no_children) == 4 + + +def test_sections_come_in_order_under_one_heading_each() -> None: + headings = [row.text for row in table_rows(_report(), NOW) if isinstance(row, Heading)] + + assert headings[0].startswith("Recorded creation failures — what the controller recorded") + assert headings[1].startswith("ResourceQuotas — current values; not a reservation") + assert headings[2].startswith("LimitRanges — current constraints; other admission plugins") + + +def test_a_failure_row_says_what_recorded_it_how_often_when_and_why() -> None: + (row, *_) = _rows(_report()) + + assert row.cells == ( + "ReplicaSet api-7d9f", + "12 times · last 2m ago", + "quota compute — quota exceeded", + EXCEEDED, + ) + assert row.destination == Destination("apps", "replicasets", "shop", "api-7d9f", "uid-rs") + + +def test_a_condition_failure_and_an_undated_one_are_labelled() -> None: + condition = replace( + FAILURE, subject=DEPLOYMENT, source=FailureSource.CONDITION, count=1, at=None + ) + + (row, *_) = _rows(_report(failures=(condition,))) + + assert row.cells[:2] == ("Deployment api (condition)", "once · undated") + assert row.destination == Destination("apps", "deployments", "shop", "api", "uid-api") + + +@pytest.mark.parametrize( + ("quotas", "observation", "expected"), + [ + ((), Basis.OBSERVED, "quota compute (no longer present) — quota exceeded"), + ((), Basis.FORBIDDEN, "quota compute (not in the quotas read) — quota exceeded"), + ], +) +def test_a_quota_the_message_names_but_the_list_lacks_is_said( + quotas: tuple[QuotaProjection, ...], observation: Basis, expected: str +) -> None: + report = _report( + quotas=quotas, quotas_observation=_seen(_source("", "resourcequotas"), observation) + ) + + (row, *_) = _rows(report) + + assert row.cells[2] == expected + + +def test_limit_range_and_other_rejections_are_attributed_only_as_far_as_the_message_goes() -> None: + limit = replace( + FAILURE, + rejection=classify( + 'pods "x" is forbidden: maximum cpu usage per Container is 1, but limit is 2' + ), + ) + webhook = replace(FAILURE, rejection=classify('admission webhook "p" denied the request: no')) + + cells = [ + row.cells[2] + for row in _rows(_report(failures=(limit, webhook), quotas=(), limit_ranges=())) + ] + + assert cells == [ + "a LimitRange in this namespace — above a LimitRange maximum", + "other rejection", + ] + + +def test_quota_rows_show_values_verbatim_with_markers_and_scopes() -> None: + rows = [ + row + for row in _rows(_report(failures=())) + if row.destination and row.destination.resource == "resourcequotas" + ] + + assert [row.cells for row in rows] == [ + ("compute", "compute: requests.cpu", "used 2 / hard 2", "at its limit"), + ("", "object counts: pods", "used 20 / hard 20", "at its limit"), + ("", "object counts: services", "used 1 / hard 5", ""), + ("", "scopes", "NotTerminating; PriorityClass In gold, silver", ""), + ] + assert {row.destination for row in rows} == { + Destination("", "resourcequotas", "shop", "compute", "uid-q") + } + + +def test_a_stale_quota_says_its_status_is_not_yet_updated() -> None: + stale = replace( + QUOTA, + entries=(QuotaEntry("pods", QuotaGroup.COUNT, "20", None, Usage.NOT_ESTABLISHED),), + scopes=(), + scope_selector=(), + stale=True, + ) + + rows = [row.cells for row in _rows(_report(failures=(), quotas=(stale,), limit_ranges=()))] + + assert rows == [ + ("compute", "object counts: pods", "used ? / hard 20", "status not yet updated") + ] + + +def test_limit_range_rows_list_each_constraint() -> None: + rows = [row.cells for row in _rows(_report(failures=(), quotas=()))] + + assert rows == [("defaults", "Container cpu", "default 500m · defaultRequest 100m · max 1", "")] + + +def test_empty_policies_get_a_row_saying_so() -> None: + empty_quota = replace(QUOTA, entries=(), scopes=(), scope_selector=()) + empty_range = replace(LIMITS, values=()) + + rows = [ + row.cells + for row in _rows(_report(failures=(), quotas=(empty_quota,), limit_ranges=(empty_range,))) + ] + + assert rows == [("compute", "no hard limits", "", ""), ("defaults", "no constraints", "", "")] + + +def test_a_section_without_rows_is_left_out() -> None: + headings = [ + r + for r in table_rows(_report(failures=(), quotas=(), limit_ranges=()), NOW) + if isinstance(r, Heading) + ] + + assert headings == [] + + +def test_notes_place_the_rejections_in_time_and_counters_in_context() -> None: + notes = _plain(report_notes(_report(), NOW)) + + assert ( + "The rejections were recorded at the times shown; the quotas and LimitRanges are as read now." + in notes + ) + assert "The counters are what the controller last reported, not a count of Pods." in notes + assert not any(n.startswith(("No ResourceQuota", "No LimitRange")) for n in notes) + + +def test_no_recorded_failure_never_rules_one_out() -> None: + available = _plain(report_notes(_report(failures=()), NOW)) + denied = _plain( + report_notes( + _report( + failures=(), + events_observation=_seen(_source("events.k8s.io", "events"), Basis.FORBIDDEN), + ), + NOW, + ) + ) + + assert ( + "No creation failure is recorded in the Events and conditions read. Events expire, so " + "this does not show that none happened." + ) in available + assert any( + n.startswith("Not established: whether a creation failure is recorded: your permissions") + for n in denied + ) + + +def test_absent_policies_are_said_only_from_a_conclusive_list() -> None: + empty = _plain(report_notes(_report(quotas=(), limit_ranges=()), NOW)) + truncated = _plain( + report_notes( + _report( + quotas=(), + quotas_observation=_seen(_source("", "resourcequotas"), Basis.TRUNCATED), + limit_ranges=(), + limit_ranges_observation=_seen(_source("", "limitranges"), Basis.FORBIDDEN), + ), + NOW, + ) + ) + + assert "No ResourceQuota in namespace shop." in empty + assert "No LimitRange in namespace shop." in empty + assert not any( + n.startswith("No ResourceQuota") or n.startswith("No LimitRange") for n in truncated + ) + assert ( + "Not established: whether every ResourceQuota is listed: stopped at the page limit" + in truncated + ) + assert any( + n.startswith("Not established: whether every LimitRange is listed: your permissions") + for n in truncated + ) + + +def test_a_stale_policy_list_no_longer_rules_anything_out() -> None: + later = NOW + timedelta(minutes=5) + + notes = _plain(report_notes(_report(quotas=(), limit_ranges=()), later)) + + assert "Not established: whether every ResourceQuota is listed: the list is stale" in notes + assert "Not established: whether every LimitRange is listed: the list is stale" in notes + + +def test_an_inconclusive_child_list_is_said() -> None: + report = _report(children_observation=_seen(_source("apps", "replicasets"), Basis.FORBIDDEN)) + + notes = _plain(report_notes(report, NOW)) + + assert any( + n.startswith("Not established: whether every ReplicaSet is listed: your permissions") + for n in notes + ) + + +def test_earlier_incarnations_are_counted() -> None: + notes = _plain(report_notes(_report(earlier_failures=2), NOW)) + + assert ( + "2 creation failures recorded about earlier incarnations of this Deployment or its " + "ReplicaSets are not shown." + ) in notes + job = _report( + target=CreationTarget("Job", "shop", "j"), earlier_failures=1, children_observation=None + ) + assert ( + "1 creation failures recorded about earlier incarnations of this Job are not shown." + ) in _plain(report_notes(job, NOW)) + + +def test_a_gone_or_recreated_workload_is_said() -> None: + gone = _report( + workload=None, + failures=(), + workload_observation=_seen(WORKLOAD_SOURCE, Basis.OBJECT_ABSENT), + ) + recreated = _report(workload_observation=_seen(WORKLOAD_SOURCE, uid="uid-new")) + + assert "The Deployment no longer exists." in _plain(report_notes(gone, NOW)) + assert not any("counters" in n for n in _plain(report_notes(gone, NOW))) + assert ( + "The Deployment did not exist when it was read; it may have been created since." + in _plain(report_notes(gone, NOW + timedelta(minutes=5))) + ) + assert ( + "The Deployment was recreated since it was selected: only the new one's evidence is shown." + ) in _plain(report_notes(recreated, NOW)) + + +class _Host(App[None]): + def __init__(self, screen: CreationConstraintsScreen) -> None: + super().__init__() + self.prompt = screen + self.result: list[Destination | None] = [] + + def on_mount(self) -> None: + self.push_screen(self.prompt, self.result.append) + + +async def test_enter_on_a_row_returns_where_it_leads() -> None: + app = _Host(CreationConstraintsScreen(_report(), lambda: NOW)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + table = app.screen.query_one(DataTable) + assert table.cursor_row == 1 + await pilot.press("down", "down", "enter") + await until(pilot, lambda: bool(app.result), label="view closed") + + assert app.result == [Destination("", "resourcequotas", "shop", "compute", "uid-q")] + + +async def test_enter_on_a_heading_does_nothing_and_escape_closes() -> None: + app = _Host(CreationConstraintsScreen(_report(), lambda: NOW)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + await pilot.press("up", "enter") + await pilot.pause() + assert app.result == [] + await pilot.press("escape") + await until(pilot, lambda: bool(app.result), label="view closed") + + assert app.result == [None] + + +async def test_cluster_text_is_shown_literally() -> None: + hostile = replace(FAILURE, rejection=classify("[bold]boom[/bold]")) + target = CreationTarget("Deployment", "shop", "[red]api[/red]") + app = _Host( + CreationConstraintsScreen( + _report(target=target, failures=(hostile,), workload=None), lambda: NOW + ) + ) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + title = app.screen.query_one("#creation-title", Static) + assert "[red]api[/red]" in str(title.render()) + table = app.screen.query_one(DataTable) + assert str(table.get_row_at(1)[3]) == "[bold]boom[/bold]" + + +async def test_an_open_view_ages_its_evidence_without_reading_again() -> None: + now = [NOW] + app = _Host(CreationConstraintsScreen(_report(quotas=()), lambda: now[0], refresh_seconds=0.05)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + evidence = app.screen.query_one("#creation-evidence", Static) + notes = app.screen.query_one("#creation-notes", Static) + assert "No ResourceQuota in namespace shop." in str(notes.render()) + + now[0] = NOW + timedelta(minutes=5) + await until(pilot, lambda: "· stale" in str(evidence.render()), label="evidence aged") + + assert "No ResourceQuota in namespace shop." not in str(notes.render()) + + +def test_without_the_workload_there_are_no_counters_to_explain() -> None: + denied = _report(workload=None, workload_observation=_seen(WORKLOAD_SOURCE, Basis.FORBIDDEN)) + + assert not any("counters" in n for n in _plain(report_notes(denied, NOW))) + + +def test_a_policy_that_could_not_be_read_in_full_never_claims_to_be_empty() -> None: + quota = replace(QUOTA, entries=(), scopes=(), scope_selector=(), complete=False) + partial = replace(QUOTA, name="partial", scopes=(), scope_selector=(), complete=False) + limit_range = replace(LIMITS, values=(), complete=False) + + rows = [ + row.cells + for row in _rows(_report(failures=(), quotas=(quota, partial), limit_ranges=(limit_range,))) + ] + + assert ("compute", "constraints could not be read", "", "") in rows + assert ("", "constraints could not be read", "", "") in rows + assert ("defaults", "constraints could not be read", "", "") in rows + assert not any(cells[1] in ("no hard limits", "no constraints") for cells in rows) + + +async def test_an_open_view_ages_its_table_cells_too() -> None: + now = [NOW] + report = _report(quotas=(), limit_ranges=()) + app = _Host(CreationConstraintsScreen(report, lambda: now[0], refresh_seconds=0.05)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + table = app.screen.query_one(DataTable) + assert str(table.get_row_at(1)[1]) == "12 times · last 2m ago" + assert str(table.get_row_at(1)[2]) == "quota compute (no longer present) — quota exceeded" + + now[0] = NOW + timedelta(minutes=5) + await until( + pilot, + lambda: str(table.get_row_at(1)[1]) == "12 times · last 7m ago", + label="age refreshed", + ) + + assert str(table.get_row_at(1)[2]) == ( + "quota compute (not in the quotas read) — quota exceeded" + ) + assert table.cursor_row == 1 + + +async def test_a_multiline_message_is_shown_in_full() -> None: + multiline = replace(FAILURE, rejection=classify("first line\nsecond line\nthird line")) + app = _Host( + CreationConstraintsScreen( + _report(failures=(multiline,), quotas=(), limit_ranges=()), lambda: NOW + ) + ) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + table = app.screen.query_one(DataTable) + row_key = table.coordinate_to_cell_key(Coordinate(1, 0)).row_key + assert table.rows[row_key].height == 3 + assert str(table.get_row_at(1)[3]) == "first line\nsecond line\nthird line" + + +def test_an_unestablished_comparison_is_marked_not_left_blank() -> None: + unknown = replace( + QUOTA, + entries=(QuotaEntry("pods", QuotaGroup.COUNT, "20", "lots", Usage.NOT_ESTABLISHED),), + scopes=(), + scope_selector=(), + ) + + rows = [row.cells for row in _rows(_report(failures=(), quotas=(unknown,), limit_ranges=()))] + + assert rows == [("compute", "object counts: pods", "used lots / hard 20", "not established")] + + +def test_no_recorded_failure_names_only_the_sources_actually_read() -> None: + unread_children = _report( + failures=(), children_observation=_seen(_source("apps", "replicasets"), Basis.FORBIDDEN) + ) + unread_workload = _report( + failures=(), + workload=None, + workload_observation=_seen(WORKLOAD_SOURCE, Basis.FORBIDDEN), + ) + + for report in (unread_children, unread_workload): + notes = _plain(report_notes(report, NOW)) + assert ( + "No creation failure is recorded in the Events read. Events expire, so this does " + "not show that none happened." + ) in notes + assert ( + "Not established: whether a failure is recorded about the Deployment or its " + "ReplicaSets: they could not all be read" + ) in notes + assert not any("Events and conditions read" in n for n in notes) + + +def test_a_kind_without_replica_failure_conditions_names_only_events() -> None: + job = _report(target=CreationTarget("Job", "shop", "j"), failures=(), children_observation=None) + + assert ( + "No creation failure is recorded in the Events read. Events expire, so this does not " + "show that none happened." + ) in _plain(report_notes(job, NOW)) +``` + +Create `src/korvid/ui/widgets/creation_constraints_screen.py` with: + +```python +"""Why a workload's Pods are not being created: recorded failures and namespace policies (#411). + +`CreationConstraintsScreen` shows one `CreationReport`: the workload and its +counters, a line per evidence source from the shared renderer, then the +recorded creation failures, the namespace's ResourceQuotas and its +LimitRanges, and what the evidence does not establish. It performs no I/O, +but re-renders the evidence lines and notes against its clock while it is +open, so a policy list that ages past its freshness policy stops ruling +anything out. Enter on a row dismisses with the `Destination` it leads to; +Escape dismisses with `None`. + +Every cell is a plain `Text`, so cluster-provided names and messages are +shown literally. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from typing import ClassVar + +from rich.text import Text +from textual.app import ComposeResult +from textual.binding import Binding +from textual.screen import ModalScreen +from textual.widgets import DataTable, Footer, Static +from textual.widgets.data_table import ColumnKey + +from korvid.core.capability import Basis, EvidenceState, Observation, conclusive, is_fresh +from korvid.core.creation_constraints import CreationReport, declarations +from korvid.core.creation_failures import Failure, FailureSource, RejectionKind +from korvid.core.creation_policies import LimitRangeProjection, LimitValue, QuotaProjection +from korvid.core.creation_quantities import Usage +from korvid.core.creation_workloads import WORKLOAD_KINDS +from korvid.ui.evidence_text import ( + basis_sentence, + elapsed_text, + not_established, + observation_line, +) + +_COLUMNS = ("Object", "What", "Value", "Note") +_UNREADABLE = ("", "constraints could not be read", "", "") +#: The kinds whose own status can record a creation failure, as a ReplicaFailure condition. +_CONDITION_KINDS = frozenset({"Deployment", "ReplicaSet"}) +_FAILURES = "Recorded creation failures — what the controller recorded, when it was recorded" +_QUOTAS = "ResourceQuotas — current values; not a reservation, and the next create may still differ" +_LIMIT_RANGES = "LimitRanges — current constraints; other admission plugins can also change a Pod" +_QUOTA_KINDS = frozenset( + { + RejectionKind.QUOTA_EXCEEDED, + RejectionKind.QUOTA_REQUIRES_FIELD, + RejectionKind.QUOTA_STATUS_UNKNOWN, + } +) +_LIMIT_KINDS = frozenset( + {RejectionKind.LIMIT_MINIMUM, RejectionKind.LIMIT_MAXIMUM, RejectionKind.LIMIT_RATIO} +) +#: Each LimitRange constraint and its field name, in display order. +_LIMIT_FIELDS: tuple[tuple[str, Callable[[LimitValue], str | None]], ...] = ( + ("default", lambda value: value.default), + ("defaultRequest", lambda value: value.default_request), + ("min", lambda value: value.min), + ("max", lambda value: value.max), + ("maxLimitRequestRatio", lambda value: value.max_limit_request_ratio), +) + + +@dataclass(frozen=True, slots=True) +class Destination: + """The object a row leads to: its API group and resource, namespace, name and UID. + + The group is kept so the caller resolves the resource by identity, not by a + display alias another API may have claimed. + """ + + group: str + resource: str + namespace: str + name: str + uid: str | None = None + + +@dataclass(frozen=True, slots=True) +class Heading: + """A section heading: what the section shows, then how to read it.""" + + text: str + + +@dataclass(frozen=True, slots=True) +class Row: + """One row's four cells, and where Enter on it leads.""" + + cells: tuple[str, str, str, str] + destination: Destination | None = None + + +TableRow = Heading | Row + + +def title_text(report: CreationReport) -> str: + """The workload, and the counters its controller last reported.""" + target = report.target + title = f"{target.kind} {target.namespace}/{target.name}" + if report.workload is None: + return title + counters = " · ".join(f"{c.label} {c.value or '?'}" for c in report.workload.counters) + return f"{title} {counters}" + + +def _observations(report: CreationReport) -> dict[str, Observation | None]: + return { + "workload": report.workload_observation, + "children": report.children_observation, + "events": report.events_observation, + "quotas": report.quotas_observation, + "limit-ranges": report.limit_ranges_observation, + } + + +def evidence_lines(report: CreationReport, now: datetime) -> list[Text]: + """Each source's line, in reading order, judged by its own policy.""" + observations = _observations(report) + lines: list[Text] = [] + for declaration in declarations(report.target): + observation = observations[declaration.key] + if observation is not None: + lines.append(observation_line(observation, declaration.policy, now)) + return lines + + +def _quota_text(name: str, report: CreationReport, now: datetime) -> str: + if any(quota.name == name for quota in report.quotas): + return f"quota {name}" + if report.quotas_conclusive(now): + return f"quota {name} (no longer present)" + return f"quota {name} (not in the quotas read)" + + +def _rejection_text(failure: Failure, report: CreationReport, now: datetime) -> str: + rejection = failure.rejection + kinds = "; ".join(kind.value for kind in rejection.kinds) + targets = [_quota_text(name, report, now) for name in rejection.quotas] + if any(kind in _LIMIT_KINDS for kind in rejection.kinds): + targets.append("a LimitRange in this namespace") + return f"{', '.join(targets)} — {kinds}" if targets else kinds + + +def _failure_row(failure: Failure, report: CreationReport, now: datetime) -> Row: + subject = failure.subject + label = f"{subject.kind} {subject.name}" + if failure.source is FailureSource.CONDITION: + label = f"{label} (condition)" + when = "undated" if failure.at is None else f"last {elapsed_text(failure.at, now)}" + workload = WORKLOAD_KINDS[subject.kind] + destination = Destination( + workload.group, workload.resource, report.target.namespace, subject.name, subject.uid + ) + times = "once" if failure.count == 1 else f"{failure.count} times" + cells = (label, f"{times} · {when}", _rejection_text(failure, report, now)) + return Row((*cells, failure.rejection.message), destination) + + +def _scopes_text(quota: QuotaProjection) -> str: + parts = list(quota.scopes) + parts += [ + f"{e.scope_name} {e.operator}" + (f" {', '.join(e.values)}" if e.values else "") + for e in quota.scope_selector + ] + return "; ".join(parts) + + +def _marker(quota: QuotaProjection, usage: Usage) -> str: + if quota.stale: + return "status not yet updated" + if usage is Usage.NOT_ESTABLISHED: + return "not established" + return "at its limit" if usage is Usage.AT_LIMIT else "" + + +def _quota_rows(quota: QuotaProjection, namespace: str) -> list[Row]: + destination = Destination("", "resourcequotas", namespace, quota.name, quota.uid) + cells: list[tuple[str, str, str, str]] = [ + ( + "", + f"{entry.group.value}: {entry.resource}", + f"used {entry.used or '?'} / hard {entry.hard}", + _marker(quota, entry.usage), + ) + for entry in quota.entries + ] + if not quota.complete: + cells.append(_UNREADABLE) + elif not cells: + cells.append(("", "no hard limits", "", "")) + scopes = _scopes_text(quota) + if scopes: + cells.append(("", "scopes", scopes, "")) + cells[0] = (quota.name, *cells[0][1:]) + return [Row(row, destination) for row in cells] + + +def _limit_text(value: LimitValue) -> str: + return " · ".join( + f"{name} {quantity}" + for name, read in _LIMIT_FIELDS + if (quantity := read(value)) is not None + ) + + +def _limit_range_rows(limit_range: LimitRangeProjection, namespace: str) -> list[Row]: + destination = Destination("", "limitranges", namespace, limit_range.name, limit_range.uid) + cells: list[tuple[str, str, str, str]] = [ + ("", f"{value.type} {value.resource}", _limit_text(value), "") + for value in limit_range.values + ] + if not limit_range.complete: + cells.append(_UNREADABLE) + elif not cells: + cells.append(("", "no constraints", "", "")) + cells[0] = (limit_range.name, *cells[0][1:]) + return [Row(row, destination) for row in cells] + + +def table_rows(report: CreationReport, now: datetime) -> list[TableRow]: + """The failures, quotas and LimitRanges under their headings; an empty section is left out.""" + namespace = report.target.namespace + sections: list[tuple[str, list[Row]]] = [ + (_FAILURES, [_failure_row(f, report, now) for f in report.failures]), + (_QUOTAS, [row for q in report.quotas for row in _quota_rows(q, namespace)]), + ( + _LIMIT_RANGES, + [row for r in report.limit_ranges for row in _limit_range_rows(r, namespace)], + ), + ] + rows: list[TableRow] = [] + for heading, section in sections: + if section: + rows.append(Heading(heading)) + rows += section + return rows + + +def _workload_notes(report: CreationReport, now: datetime) -> list[Text]: + kind = report.target.kind + observation = report.workload_observation + if observation.basis is Basis.OBJECT_ABSENT: + if not is_fresh(observation, declarations(report.target)[0].policy, now): + return [ + Text(f"The {kind} did not exist when it was read; it may have been created since.") + ] + return [Text(f"The {kind} no longer exists.")] + notes: list[Text] = [] + if report.recreated: + notes.append( + Text( + f"The {kind} was recreated since it was selected: only the new one's " + "evidence is shown." + ) + ) + if report.workload is not None: + notes.append( + Text("The counters are what the controller last reported, not a count of Pods.") + ) + return notes + + +def _failure_notes(report: CreationReport) -> list[Text]: + target = report.target + notes: list[Text] = [] + if report.earlier_failures: + children = f" or its {target.child}s" if target.child is not None else "" + notes.append( + Text( + f"{report.earlier_failures} creation failures recorded about earlier " + f"incarnations of this {target.kind}{children} are not shown." + ) + ) + if report.failures: + notes.append( + Text( + "The rejections were recorded at the times shown; the quotas and LimitRanges " + "are as read now." + ) + ) + elif report.events_observation.state is EvidenceState.AVAILABLE: + notes += _nothing_recorded(report) + else: + reason = basis_sentence(report.events_observation.basis) + notes.append(not_established(f"whether a creation failure is recorded: {reason}")) + return notes + + +def _nothing_recorded(report: CreationReport) -> list[Text]: + """No failure in what was read: never a conclusion, and only about what was read.""" + target = report.target + children = report.children_observation + subjects_read = report.workload_observation.state is EvidenceState.AVAILABLE and ( + children is None or children.state is EvidenceState.AVAILABLE + ) + sources = ( + "Events and conditions" if subjects_read and target.kind in _CONDITION_KINDS else "Events" + ) + notes = [ + Text( + f"No creation failure is recorded in the {sources} read. Events expire, so this " + "does not show that none happened." + ) + ] + if not subjects_read: + about = target.kind + (f" or its {target.child}s" if target.child is not None else "") + notes.append( + not_established( + f"whether a failure is recorded about the {about}: they could not all be read" + ) + ) + return notes + + +def _listed( + report: CreationReport, key: str, observation: Observation | None, now: datetime +) -> bool | str: + """`True` when the list is conclusive, otherwise why it is not.""" + if observation is None: + return True + policy = next(d.policy for d in declarations(report.target) if d.key == key) + if conclusive(observation, policy, now): + return True + if observation.state is EvidenceState.AVAILABLE and not is_fresh(observation, policy, now): + return "the list is stale" + return basis_sentence(observation.basis) + + +def _completeness_notes(report: CreationReport, now: datetime) -> list[Text]: + namespace = report.target.namespace + notes: list[Text] = [] + child = report.target.child + children = _listed(report, "children", report.children_observation, now) + if child is not None and children is not True: + notes.append(not_established(f"whether every {child} is listed: {children}")) + for key, observation, items, kind in ( + ("quotas", report.quotas_observation, report.quotas, "ResourceQuota"), + ("limit-ranges", report.limit_ranges_observation, report.limit_ranges, "LimitRange"), + ): + listed = _listed(report, key, observation, now) + if listed is not True: + notes.append(not_established(f"whether every {kind} is listed: {listed}")) + elif not items: + notes.append(Text(f"No {kind} in namespace {namespace}.")) + return notes + + +def report_notes(report: CreationReport, now: datetime) -> list[Text]: + """What the evidence says about the workload, and what it does not establish.""" + return [ + *_workload_notes(report, now), + *_failure_notes(report), + *_completeness_notes(report, now), + ] + + +class CreationConstraintsScreen(ModalScreen[Destination | None]): + """One workload's recorded creation failures and its namespace's policies.""" + + BINDINGS: ClassVar[list[Binding | tuple[str, str] | tuple[str, str, str]]] = [ + Binding("escape", "close", "Close", show=True), + Binding("q", "close", "Close", show=False), + ] + + DEFAULT_CSS = """ + CreationConstraintsScreen { + layout: vertical; + background: $background; + } + CreationConstraintsScreen #creation-title { + padding: 0 1; + text-style: bold; + } + CreationConstraintsScreen #creation-evidence, CreationConstraintsScreen #creation-notes { + padding: 0 1; + color: $text-muted; + } + CreationConstraintsScreen DataTable { + height: 1fr; + } + """ + + def __init__( + self, + report: CreationReport, + clock: Callable[[], datetime], + *, + refresh_seconds: float = 5.0, + ) -> None: + super().__init__() + self.report = report + self._clock = clock + self._refresh_seconds = refresh_seconds + self._rows: dict[str, Destination] = {} + self._columns: list[ColumnKey] = [] + + def compose(self) -> ComposeResult: + yield Footer() + yield Static(title_text(self.report), id="creation-title", markup=False) + yield Static(id="creation-evidence") + yield DataTable[Text](id="creation-table") + yield Static(id="creation-notes") + + def _render_text(self) -> None: + now = self._clock() + evidence = Text("\n").join(evidence_lines(self.report, now)) + self.query_one("#creation-evidence", Static).update(evidence) + notes = Text("\n").join(report_notes(self.report, now)) + self.query_one("#creation-notes", Static).update(notes) + self._refresh_rows(now) + + def _refresh_rows(self, now: datetime) -> None: + """Re-render cells that depend on the time: ages, and whether a list still rules out.""" + if not self._columns: + return + table: DataTable[Text] = self.query_one(DataTable) + for index, row in enumerate(table_rows(self.report, now)): + if isinstance(row, Row): + for column, cell in zip(self._columns, row.cells, strict=True): + table.update_cell(f"r{index}", column, Text(cell)) + + def on_mount(self) -> None: + self._render_text() + self.set_interval(self._refresh_seconds, self._render_text) + table: DataTable[Text] = self.query_one(DataTable) + table.cursor_type = "row" + self._columns = table.add_columns(*_COLUMNS) + first: int | None = None + for index, row in enumerate(table_rows(self.report, self._clock())): + if isinstance(row, Heading): + title, _, meaning = row.text.partition(" — ") + table.add_row(Text(title, style="bold"), Text(""), Text(meaning), key=f"h{index}") + continue + key = f"r{index}" + if row.destination is not None: + self._rows[key] = row.destination + # A rejection message can span lines; the row grows to show all of it. + table.add_row(*(Text(cell) for cell in row.cells), key=key, height=None) + first = index if first is None else first + if first is not None: + table.move_cursor(row=first) + table.focus() + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + """Enter on a row returns where it leads; a heading does nothing.""" + destination = self._rows.get(str(event.row_key.value)) + if destination is not None: + self.dismiss(destination) + + def action_close(self) -> None: + self.dismiss(None) +``` + +- [ ] Add `elapsed_text` to `src/korvid/ui/evidence_text.py` (`2m ago`, `just now`, `in the future`), and make `age_text` prefix it with `observed`; test it in `tests/ui/test_evidence_text.py`. +- [ ] Run `uv run --frozen pytest -p no:tach tests/ui/test_creation_constraints_screen.py tests/ui/test_evidence_text.py -q` before (collection error) and after (PASS), then commit: `Show a workload's creation failures and namespace policies in a view`. + +### Task 2: The controller + +Create `tests/ui/test_creation_constraints_controller.py` with: + +```python +"""The creation constraints controller: availability, runs and late results (#411).""" + +from __future__ import annotations + +from collections.abc import Awaitable, Mapping +from datetime import UTC, datetime +from typing import Any + +from korvid.core.creation_constraints import CreationTarget +from korvid.core.creation_failures import Subject +from korvid.k8s.discovery import ResourceMeta +from korvid.k8s.discovery_snapshot import ( + GroupVersionOutcome, + RawDiscovery, + RootOutcome, + ServedResource, +) +from korvid.k8s.evidence_requests import EvidenceClient, EvidencePage, EvidenceTarget +from korvid.ui.action_availability import AvailabilityCode +from korvid.ui.creation_constraints_controller import ( + CREATION_CONSTRAINTS_GROUP, + CreationConstraintsController, +) +from korvid.ui.view_state import ViewState +from korvid.ui.widgets.creation_constraints_screen import CreationConstraintsScreen, Destination +from korvid.ui.workspace_controller import ContextGuard + +from .test_session_timeline_controller import FakeUiSurface + +NOW = datetime(2026, 10, 30, 10, 0, tzinfo=UTC) +DEPLOYMENTS = ResourceMeta("Deployment", "deployments", "apps", "v1", True) +CRON_JOBS = ResourceMeta("CronJob", "cronjobs", "batch", "v1", True) +PODS = ResourceMeta("Pod", "pods", "", "v1", True) +QUOTAS = ResourceMeta("ResourceQuota", "resourcequotas", "", "v1", True) +REPLICA_SETS = ResourceMeta("ReplicaSet", "replicasets", "apps", "v1", True) +FOREIGN_REPLICA_SETS = ResourceMeta("ReplicaSet", "replicasets", "vendor.example.com", "v1", True) +FOREIGN = ResourceMeta("Deployment", "deployments", "vendor.example.com", "v1", True) +_SERVED = { + ("", "v1"): ("resourcequotas", "limitranges", "events"), + ("apps", "v1"): ("deployments", "replicasets"), + ("batch", "v1"): ("jobs", "cronjobs"), +} +EVENT = { + "metadata": {"name": "e", "namespace": "shop", "uid": "uid-e"}, + "reason": "FailedCreate", + "involvedObject": {"kind": "ReplicaSet", "name": "api-7d9f", "uid": "uid-rs"}, + "message": 'pods "x" is forbidden: exceeded quota: compute, requested: pods=1', +} + + +class FakeView(ViewState): + def __init__(self, kind: str = "deployments", selected: tuple[str, str] | None = None) -> None: + self.kind = kind + self.scope = "shop" + self.selected = selected + self.alias_map = { + "deployments": DEPLOYMENTS, + "cronjobs": CRON_JOBS, + "pods": PODS, + "deployments.vendor.example.com": FOREIGN, + "resourcequotas": QUOTAS, + } + + def current_kind(self) -> str: + return self.kind + + def canonical_kind(self, kind: str) -> str: + return kind + + def aliases(self) -> Mapping[str, ResourceMeta]: + return self.alias_map + + def selected_ns_name(self, *, notify: bool = True) -> tuple[str | None, str | None]: + return self.selected if self.selected is not None else (None, None) + + def selected_uid(self, namespace: str | None, name: str) -> str | None: + return f"uid-{name}" + + def current_scope(self) -> str: + return self.scope + + def resources(self, kind: str, scope: str) -> list[Any]: + return [] + + def readonly(self) -> bool: + return False + + def default_namespace(self) -> str | None: + return "shop" + + def gvr_label(self, meta: ResourceMeta) -> str: + return meta.plural + + def write_locus(self, namespace: str | None) -> str: + return namespace or "" + + +class FakeContext(ContextGuard): + def __init__(self) -> None: + self.current = 1 + self.in_switch = False + self.allowed = True + + def epoch(self) -> int: + return self.current + + def switching(self) -> bool: + return self.in_switch + + def reads_allowed(self) -> bool: + return self.allowed + + def crossed(self, epoch: int) -> bool: + return self.in_switch or epoch != self.current + + +class FakeClient(EvidenceClient): + def __init__(self) -> None: + self.generation = 3 + self.gets: list[tuple[EvidenceTarget, str]] = [] + self.lists: list[EvidenceTarget] = [] + + @property + def session_generation(self) -> int: + return self.generation + + async def discovery_snapshot(self) -> RawDiscovery: + outcomes = tuple( + GroupVersionOutcome( + group, version, tuple(ServedResource(r, frozenset({"get", "list"})) for r in names) + ) + for (group, version), names in _SERVED.items() + ) + groups = tuple((g, v) for g, v in _SERVED if g) + return RawDiscovery( + self.generation, + NOW, + 0.0, + RootOutcome("/api", (("", "v1"),)), + RootOutcome("/apis", groups), + outcomes, + ) + + async def list_evidence_page( + self, target: EvidenceTarget, *, limit: int, continuation: str, max_bytes: int + ) -> EvidencePage: + self.lists.append(target) + if target.resource == "replicasets": + owner = [{"uid": "uid-api", "controller": True}] + rs = { + "metadata": { + "name": "api-7d9f", + "namespace": "shop", + "uid": "uid-rs", + "ownerReferences": owner, + } + } + return EvidencePage((rs,)) + if target.resource == "events": + return EvidencePage((EVENT,)) + return EvidencePage(()) + + async def get_evidence_object( + self, target: EvidenceTarget, name: str, *, max_bytes: int + ) -> dict[str, Any]: + self.gets.append((target, name)) + return { + "metadata": {"name": name, "namespace": "shop", "uid": f"uid-{name}"}, + "spec": {}, + "status": {}, + } + + +class Harness: + def __init__(self, *, client: FakeClient | None = None, view: FakeView | None = None) -> None: + self.ui = FakeUiSurface(close_worker_coroutines=False) + self.view = view or FakeView(selected=("shop", "api")) + self.context = FakeContext() + self.client = FakeClient() if client is None else client + self.navigations: list[tuple[str, str, str, int, str | None]] = [] + self.controller = CreationConstraintsController( + ui=self.ui, + view=self.view, + context=self.context, + client=self.client, + navigate=self._navigate, + clock=lambda: NOW, + ) + + async def _navigate( + self, kind: str, namespace: str, name: str, epoch: int, uid: str | None + ) -> None: + self.navigations.append((kind, namespace, name, epoch, uid)) + + async def run(self, index: int = -1) -> None: + work = self.ui.workers[index] + assert isinstance(work, Awaitable) + await work + + +def _code(harness: Harness) -> AvailabilityCode | None: + reason = harness.controller.unavailable_reason() + return None if reason is None else reason.code + + +def test_it_is_available_on_a_selected_workload() -> None: + assert _code(Harness()) is None + assert _code(Harness(view=FakeView("cronjobs", ("shop", "nightly")))) is None + + +def test_it_says_why_it_is_unavailable() -> None: + unconnected = CreationConstraintsController( + ui=FakeUiSurface(), + view=FakeView(selected=("shop", "api")), + context=FakeContext(), + client=None, + navigate=Harness()._navigate, + ) + + reason = unconnected.unavailable_reason() + assert reason is not None + assert reason.code is AvailabilityCode.MISSING_CAPABILITY + assert _code(Harness(view=FakeView("pods", ("shop", "api-1")))) is AvailabilityCode.WRONG_VIEW + assert ( + _code(Harness(view=FakeView("deployments.vendor.example.com", ("shop", "x")))) + is AvailabilityCode.WRONG_VIEW + ) + assert _code(Harness(view=FakeView("deployments", None))) is AvailabilityCode.NO_SELECTION + + +async def test_opening_reads_the_selected_workload_and_shows_its_failures() -> None: + harness = Harness() + + harness.controller.open_selected() + await harness.run() + + assert harness.ui.worker_groups == [(CREATION_CONSTRAINTS_GROUP, False)] + assert harness.ui.exclusive_workers == [True] + (screen,) = harness.ui.screens + assert isinstance(screen, CreationConstraintsScreen) + assert screen.report.target == CreationTarget("Deployment", "shop", "api", "uid-api") + assert [f.subject for f in screen.report.failures] == [ + Subject("ReplicaSet", "api-7d9f", "uid-rs") + ] + assert harness.client.gets == [(EvidenceTarget("apps", "v1", "deployments", "shop"), "api")] + assert {t.resource for t in harness.client.lists} == { + "replicasets", + "events", + "resourcequotas", + "limitranges", + } + + +def test_opening_where_it_is_unavailable_says_why_and_reads_nothing() -> None: + harness = Harness(view=FakeView("pods", ("shop", "api-1"))) + + harness.controller.open_selected() + + assert harness.ui.workers == [] + assert [n.severity for n in harness.ui.notifications] == ["warning"] + + +def test_opening_while_reads_are_held_reads_nothing() -> None: + harness = Harness() + harness.context.allowed = False + + harness.controller.open_selected() + + assert harness.ui.workers == [] + + +async def test_a_newer_run_drops_an_older_result() -> None: + harness = Harness() + + harness.controller.open_selected() + harness.controller.open_selected() + await harness.run(0) + + assert harness.ui.screens == [] + await harness.run(1) + assert len(harness.ui.screens) == 1 + + +async def test_a_context_switch_or_new_credentials_drop_the_result() -> None: + switched = Harness() + reauthenticated = Harness() + + switched.controller.open_selected() + reauthenticated.controller.open_selected() + switched.context.current = 2 + reauthenticated.client.generation = 4 + await switched.run() + await reauthenticated.run() + + assert (switched.ui.screens, reauthenticated.ui.screens) == ([], []) + + +async def test_choosing_a_row_opens_its_object_in_the_run_epoch() -> None: + harness = Harness() + harness.controller.open_selected() + await harness.run() + + assert harness.ui.screen_callback is not None + harness.ui.screen_callback(None) + assert len(harness.ui.workers) == 1 + harness.context.current = 2 + harness.ui.screen_callback(Destination("", "resourcequotas", "shop", "compute", "uid-q")) + await harness.run() + + assert harness.navigations == [("resourcequotas", "shop", "compute", 1, "uid-q")] + assert harness.ui.worker_groups[-1] == (CREATION_CONSTRAINTS_GROUP, False) + + +async def test_a_dialog_opened_during_the_reads_keeps_the_view_closed() -> None: + harness = Harness() + harness.controller.open_selected() + harness.ui.screen_depth = lambda: 2 # type: ignore[method-assign] # a dialog opened + + await harness.run() + + assert harness.ui.screens == [] + + +async def test_leaving_the_view_or_the_selection_keeps_the_result_closed() -> None: + moved = Harness() + rescoped = Harness() + reselected = Harness() + + for harness in (moved, rescoped, reselected): + harness.controller.open_selected() + moved.view.kind = "pods" + rescoped.view.scope = "other" + reselected.view.selected = ("shop", "other") + for harness in (moved, rescoped, reselected): + await harness.run() + + assert (moved.ui.screens, rescoped.ui.screens, reselected.ui.screens) == ([], [], []) + + +async def test_the_worker_group_is_cancelled_on_request() -> None: + harness = Harness() + + await harness.controller.cancel_workers() + + assert harness.ui.cancelled_groups == [CREATION_CONSTRAINTS_GROUP] + + +async def _choose(harness: Harness, destination: Destination) -> None: + harness.controller.open_selected() + await harness.run() + assert harness.ui.screen_callback is not None + harness.ui.screen_callback(destination) + await harness.run() + + +async def test_a_destination_is_opened_under_its_own_api_group_not_a_colliding_alias() -> None: + harness = Harness() + harness.view.alias_map["replicasets"] = FOREIGN_REPLICA_SETS + harness.view.alias_map["replicasets.apps"] = REPLICA_SETS + + await _choose(harness, Destination("apps", "replicasets", "shop", "api-7d9f", "uid-rs")) + + assert harness.navigations == [("replicasets.apps", "shop", "api-7d9f", 1, "uid-rs")] + + +async def test_a_destination_this_session_does_not_serve_is_said_not_opened() -> None: + harness = Harness() + + harness.controller.open_selected() + await harness.run() + assert harness.ui.screen_callback is not None + harness.ui.screen_callback(Destination("apps", "replicasets", "shop", "api-7d9f", "uid-rs")) + + assert harness.navigations == [] + assert [(n.message, n.severity) for n in harness.ui.notifications] == [ + ("replicasets.apps is not discovered in this session", "warning") + ] +``` + +Create `src/korvid/ui/creation_constraints_controller.py` with: + +```python +"""Open the workload creation constraints view for the selected workload (#411). + +One exclusive worker reads the workload, its children, the Events and the +namespace's quotas and LimitRanges through `explain_creation`, under a run +token. The view opens only while that token is current: a context switch, +new credentials or a newer run drops the result, as `RunTracker` decides. So +does another dialog that is open when the reads finish, or a change of view, +scope or selection since they started. The context-switch coordinator +cancels the worker group before it swaps the client. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime + +from korvid.core.api_versions import observe_discovery +from korvid.core.capability import RunToken +from korvid.core.creation_constraints import CreationTarget, explain_creation +from korvid.core.creation_workloads import WORKLOAD_KINDS +from korvid.core.run_tokens import RunTracker +from korvid.k8s.discovery import canonical_resource_alias, resolve_resource +from korvid.k8s.evidence_requests import EvidenceClient +from korvid.ui.action_availability import AvailabilityCode, UnavailableReason +from korvid.ui.ui_surface import UiSurface +from korvid.ui.view_state import ViewState +from korvid.ui.widgets.creation_constraints_screen import CreationConstraintsScreen, Destination +from korvid.ui.workspace_controller import ContextGuard + +CREATION_CONSTRAINTS_GROUP = "creation-constraints" +_UNCONNECTED = UnavailableReason( + AvailabilityCode.MISSING_CAPABILITY, "Workload creation failures need a cluster connection" +) +_WRONG_VIEW = UnavailableReason( + AvailabilityCode.WRONG_VIEW, + "Open Deployments, ReplicaSets, StatefulSets, DaemonSets, Jobs or CronJobs to explain " + "creation failures", +) +_NO_SELECTION = UnavailableReason(AvailabilityCode.NO_SELECTION, "Select a workload") + +#: Opens an object in its table: (view alias, namespace, name, epoch, expected UID). +ObjectNavigator = Callable[[str, str, str, int, str | None], Awaitable[None]] +#: The view, scope and (namespace, name) selection a run was started from. +_Origin = tuple[str, str, tuple[str | None, str | None]] + + +def _now() -> datetime: + return datetime.now(UTC) + + +class CreationConstraintsController: + """Runs the creation diagnostic and shows its result while it is current.""" + + def __init__( + self, + *, + ui: UiSurface, + view: ViewState, + context: ContextGuard, + client: EvidenceClient | None, + navigate: ObjectNavigator, + clock: Callable[[], datetime] = _now, + ) -> None: + self._ui = ui + self._view = view + self._context = context + self._client = client + self._navigate = navigate + self._clock = clock + self._tracker = RunTracker( + epoch=context.epoch, + session_generation=lambda: client.session_generation if client is not None else 0, + switching=context.switching, + ) + + def _selected_kind(self) -> str | None: + meta = self._view.aliases().get(self._view.canonical_kind(self._view.current_kind())) + workload = WORKLOAD_KINDS.get(meta.kind) if meta is not None else None + if meta is None or workload is None or meta.group != workload.group: + return None + return meta.kind + + def unavailable_reason(self) -> UnavailableReason | None: + """Why the view cannot open now, or None. + + Only for a missing connection, the wrong view or no selection: a + source the cluster refuses never disables it, the view says so. + """ + if self._client is None: + return _UNCONNECTED + if self._selected_kind() is None: + return _WRONG_VIEW + namespace, name = self._view.selected_ns_name(notify=False) + if not namespace or not name: + return _NO_SELECTION + return None + + def open_selected(self) -> None: + """Explain the selected workload's creation failures in a worker.""" + reason = self.unavailable_reason() + if reason is not None: + self._ui.notify(reason.message, severity="warning", markup=False) + return + kind = self._selected_kind() + namespace, name = self._view.selected_ns_name() + if not self._context.reads_allowed() or kind is None or not namespace or not name: + return + target = CreationTarget(kind, namespace, name, self._view.selected_uid(namespace, name)) + token = self._tracker.next_run() + self._ui.run_worker( + self._explain(target, token, self._context.epoch(), self._origin()), + exclusive=True, + group=CREATION_CONSTRAINTS_GROUP, + exit_on_error=False, + ) + + def _origin(self) -> _Origin: + """The view, scope and selected row a run explains.""" + return ( + self._view.current_kind(), + self._view.current_scope(), + self._view.selected_ns_name(notify=False), + ) + + async def _explain( + self, target: CreationTarget, token: RunToken, epoch: int, origin: _Origin + ) -> None: + client = self._client + if client is None: + return + snapshot = observe_discovery(await client.discovery_snapshot()) + report = await explain_creation(client, snapshot, target, token=token, clock=self._clock) + # Another dialog is open, or the user has moved on, as the reads finish. + if self._ui.screen_depth() > 1 or self._origin() != origin: + return + screen = CreationConstraintsScreen(report, self._clock) + self._tracker.apply_if_current( + token, + lambda: self._ui.push_screen(screen, lambda chosen: self._open(chosen, epoch)), + ) + + def _open(self, destination: Destination | None, epoch: int) -> None: + """Open a row's object by its API identity, never by a colliding display alias.""" + if destination is None: + return + aliases = self._view.aliases() + meta = resolve_resource(aliases, destination.group, destination.resource) + if meta is None: + name = destination.resource + qualified = f"{name}.{destination.group}" if destination.group else name + self._ui.notify( + f"{qualified} is not discovered in this session", severity="warning", markup=False + ) + return + alias = canonical_resource_alias(aliases, meta) + self._ui.run_worker( + self._navigate(alias, destination.namespace, destination.name, epoch, destination.uid), + group=CREATION_CONSTRAINTS_GROUP, + exit_on_error=False, + ) + + async def cancel_workers(self) -> None: + """Stop a run holding the old client, before a context switch swaps it.""" + await self._ui.cancel_workers(CREATION_CONSTRAINTS_GROUP) +``` + +- [ ] Record `exclusive` in `FakeUiSurface.run_worker` (`tests/ui/test_session_timeline_controller.py`) as `exclusive_workers`. +- [ ] Run the controller tests before and after, then commit: `Open the creation constraints view from the selected workload`. + +### Task 3: The command and the wiring + +- [ ] Add tests: `test_creation_failures_reach_their_owner_once` and `test_creation_failures_without_owner_warn_unavailable` (`tests/ui/test_command_router.py`), `test_composed_command_reasons_ask_the_creation_owner_about_blocked` (`tests/ui/test_action_policy.py`), the coordinator's cancel order (`tests/ui/test_context_switch_coordinator.py`), and `tests/ui/test_worker_failures.py`. +- [ ] Add `BuiltinOperation.CREATION_FAILURES = "blocked"`; a `CommandDescriptor` with help `Pods not created for a workload: recorded failures, quotas, LimitRanges` and palette entry `Explain workload creation failures`; `CreationCommands` and `_open_creation` in the router; `creation` in `compose_command_reasons` (key `blocked`); `cancel_worker_group(CREATION_CONSTRAINTS_GROUP)` after the propagation group; the worker-failure label `Workload creation failure explanation failed`; and the controller in `__main__.py`, navigating with `jump_to_object(kind, namespace, name, epoch=epoch, expected_uid=uid)`. +- [ ] Run the router, policy, coordinator, wiring and palette tests, then commit: `Wire :blocked to the creation constraints view`. + +### Task 4: End to end + +Create `tests/ui/test_creation_constraints_workflow.py` with: + +```python +"""`:blocked` end to end: a Deployment row, its recorded failures, and their ReplicaSet (#411).""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from datetime import UTC, datetime +from typing import Any + +from korvid.core.config import KorvidConfig +from korvid.core.store import ResourceStore, Summary +from korvid.core.watch import WatchManager +from korvid.k8s.discovery import PODS_META, ResourceMeta, build_alias_map +from korvid.k8s.discovery_snapshot import ( + GroupVersionOutcome, + RawDiscovery, + RootOutcome, + ServedResource, +) +from korvid.k8s.evidence_requests import EvidenceClient, EvidencePage, EvidenceTarget +from korvid.k8s.models import GenericSummary +from korvid.ui.widgets.creation_constraints_screen import CreationConstraintsScreen +from tests.app_factory import build_test_app +from tests.ui.waits import screen_ready, until + +DEPLOYMENTS = ResourceMeta("Deployment", "deployments", "apps", "v1", True) +REPLICA_SETS = ResourceMeta("ReplicaSet", "replicasets", "apps", "v1", True) +EXCEEDED = 'pods "api-7d9f-x" is forbidden: exceeded quota: compute, requested: pods=1' +_SERVED = { + ("", "v1"): ("resourcequotas", "limitranges", "events"), + ("apps", "v1"): ("deployments", "replicasets"), +} + + +class Client(EvidenceClient): + """Serves one Deployment whose ReplicaSet records a quota rejection.""" + + def __init__(self) -> None: + self.reads: list[tuple[str, str, str]] = [] + + @property + def session_generation(self) -> int: + return 1 + + async def discovery_snapshot(self) -> RawDiscovery: + outcomes = tuple( + GroupVersionOutcome( + group, version, tuple(ServedResource(r, frozenset({"get", "list"})) for r in names) + ) + for (group, version), names in _SERVED.items() + ) + return RawDiscovery( + 1, + datetime.now(UTC), + 0.0, + RootOutcome("/api", (("", "v1"),)), + RootOutcome("/apis", (("apps", "v1"),)), + outcomes, + ) + + async def list_evidence_page( + self, target: EvidenceTarget, *, limit: int, continuation: str, max_bytes: int + ) -> EvidencePage: + self.reads.append(("list", target.resource, target.namespace or "")) + if target.resource == "replicasets": + owner = [{"uid": "uid-api", "controller": True}] + rs = { + "metadata": { + "name": "api-7d9f", + "namespace": "default", + "uid": "uid-rs", + "ownerReferences": owner, + } + } + return EvidencePage((rs,)) + if target.resource == "events": + event = { + "metadata": {"name": "e", "namespace": "default", "uid": "uid-e"}, + "reason": "FailedCreate", + "involvedObject": {"kind": "ReplicaSet", "name": "api-7d9f", "uid": "uid-rs"}, + "message": EXCEEDED, + } + return EvidencePage((event,)) + if target.resource == "resourcequotas": + quota = { + "metadata": {"name": "compute", "namespace": "default", "uid": "uid-q"}, + "spec": {"hard": {"pods": "0"}}, + "status": {"hard": {"pods": "0"}, "used": {"pods": "0"}}, + } + return EvidencePage((quota,)) + return EvidencePage(()) + + async def get_evidence_object( + self, target: EvidenceTarget, name: str, *, max_bytes: int + ) -> dict[str, Any]: + self.reads.append(("get", target.resource, name)) + template = { + "spec": { + "containers": [{"name": "app", "env": [{"name": "T", "value": "s3cr3t-literal"}]}] + } + } + return { + "metadata": {"name": name, "namespace": "default", "uid": "uid-api"}, + "spec": {"replicas": 1, "template": template}, + "status": {}, + } + + +async def _source(kind: str, scope: str) -> AsyncIterator[tuple[str, Summary]]: + if kind == "deployments": + yield ("ADDED", GenericSummary("api", "default", "Deployment", "", uid="uid-api")) + if kind == "replicasets": + yield ("ADDED", GenericSummary("api-7d9f", "default", "ReplicaSet", "", uid="uid-rs")) + await asyncio.Event().wait() + + +async def _command(pilot: Any, text: str) -> None: + await pilot.press("colon", *text, "enter") + + +def _app(client: EvidenceClient) -> Any: + store = ResourceStore() + return build_test_app( + config=KorvidConfig(namespace="default"), + store=store, + watch_manager=WatchManager(store, _source), + aliases=build_alias_map([PODS_META, DEPLOYMENTS, REPLICA_SETS]), + agent_available=False, + evidence_client=client, + ) + + +async def test_a_deployments_failures_lead_to_their_replica_set() -> None: + client = Client() + app = _app(client) + async with app.run_test() as pilot: + await _command(pilot, "deployments") + await until( + pilot, + lambda: app._view.selected_ns_name(notify=False) == ("default", "api"), + label="Deployment selected", + ) + + await _command(pilot, "blocked") + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + screen = app.screen + assert isinstance(screen, CreationConstraintsScreen) + (failure,) = screen.report.failures + assert failure.rejection.quotas == ("compute",) + assert "s3cr3t-literal" not in repr(screen.report) + assert sorted(client.reads) == [ + ("get", "deployments", "api"), + ("list", "events", "default"), + ("list", "limitranges", "default"), + ("list", "replicasets", "default"), + ("list", "resourcequotas", "default"), + ] + + await pilot.press("enter") + await until(pilot, lambda: app.current_kind == "replicasets", label="ReplicaSet view") + await until( + pilot, + lambda: app._view.selected_ns_name(notify=False) == ("default", "api-7d9f"), + label="ReplicaSet selected", + ) + + +async def test_blocked_on_another_view_says_what_to_open() -> None: + app = _app(Client()) + async with app.run_test() as pilot: + await _command(pilot, "blocked") + await until( + pilot, + lambda: any("Open Deployments, ReplicaSets" in n.message for n in app._notifications), + label="reason shown", + ) + assert not isinstance(app.screen, CreationConstraintsScreen) + + +class BrokenClient(Client): + async def discovery_snapshot(self) -> RawDiscovery: + raise RuntimeError("discovery exploded") + + +async def test_a_failed_run_is_reported_instead_of_vanishing() -> None: + app = _app(BrokenClient()) + async with app.run_test() as pilot: + await _command(pilot, "deployments") + await until( + pilot, + lambda: app._view.selected_ns_name(notify=False) == ("default", "api"), + label="Deployment selected", + ) + await _command(pilot, "blocked") + await until( + pilot, + lambda: any( + "Workload creation failure explanation failed" in n.message + and "discovery exploded" in n.message + for n in app._notifications + ), + label="failure reported", + ) + assert app.is_running +``` + +- [ ] Run it, then commit: `Cover :blocked end to end`. + +### Task 5: The version matrix + +- [ ] In `tests/version_matrix/fixtures.yaml`, add a `korvid-matrix-quota` namespace containing: + - a `pods: "0"` ResourceQuota `matrix-pods`; + - a Container LimitRange `matrix-limits`; + - a one-replica Deployment `blocked` whose template carries `matrix-template-literal`. + + It is its own namespace, so the quota does not block the other fixtures. +- [ ] Add `test_a_quota_rejection_is_read_next_to_the_namespace_policies` to `tests/version_matrix/test_scenarios.py`. It polls `explain_creation` until a failure is recorded and the quota's status is current, up to 90 seconds. +- [ ] Add the `creation_*` modules and `k8s/events.py` to the workflow's path filters, then commit: `Add a version-matrix scenario for workload creation constraints`. + +### Task 6: Document it + +- [ ] Add a "Workload creation failures" section to `docs/tui.md`, the `:blocked` row to `docs/keybindings.md`, the scenario to `docs/kubernetes-support.md` and an entry to `docs/release-notes/unreleased.md`, then commit: `Document :blocked`. + +### Task 7: Run the full gate + +- [ ] Run `make check`, or behind the mirror the same chain with the venv's tools: `scripts/check_source_size.py`, `ruff check`, `ruff format --check`, `mypy src/`, `tach check`, `pytest`. +- [ ] Mutation-check each branch of the screen's rows and notes and the controller's guards. The mutation helper runs the unmutated tests first and refuses to report on a failing baseline. diff --git a/docs/tui.md b/docs/tui.md index 3314daf96..0cab11fab 100644 --- a/docs/tui.md +++ b/docs/tui.md @@ -168,6 +168,46 @@ These limits apply: - korvid keeps no history across restarts and never reads `creationTimestamp` as a change time. +### Workload creation failures + +A Deployment can exist, or be scaled up, while its Pods are rejected before +they are created: the Pod list is empty, so nothing on it explains why. +Select a Deployment, ReplicaSet, StatefulSet, DaemonSet, Job or CronJob and +run `:blocked` (or **Explain workload creation failures** in the Action +Palette). The view shows the failures recorded for that workload next to its +namespace's ResourceQuotas and LimitRanges: + +| Section | What it shows | +|---|---| +| Header | The workload and the counters its controller last reported, such as desired and created replicas | +| Recorded creation failures | Each `FailedCreate` event about the workload or its ReplicaSets or Jobs, and each `ReplicaFailure` condition, with how often and when it was last recorded, and what the rejection says | +| ResourceQuotas | Each quota's `used / hard` values, grouped as compute, storage, object counts and other, with a resource at or over its limit marked, and its scopes | +| LimitRanges | Each constraint: defaults, minimums, maximums and limit-to-request ratios | + +How to read it: + +- **A rejection is attributed only as far as its message goes.** A message + that names a quota links to it. A LimitRange message is attributed to "a + LimitRange in this namespace", because LimitRanger messages do not say + which. A rejection an admission webhook or a ValidatingAdmissionPolicy + wrote is shown verbatim and attributed to nothing. +- **Quantities are shown as the cluster wrote them**, and compared exactly: + `1Gi` is never rounded or converted, and `0.5` equals `500m`. +- **Times are kept apart.** A rejection was recorded at its own time; the + quotas are as read now. Quota status is not a reservation, and the next + create may still differ. +- **Absence is said only when it is a fact.** "No ResourceQuota" appears + only when every quota could be listed. A quota whose status has not caught + up with its spec is marked, and is never called "at its limit". Events + expire, so finding no recorded failure does not show that none happened. + +The view reads the workload, its ReplicaSets or Jobs, the namespace's +Events, ResourceQuotas and LimitRanges when it opens, and nothing else. The +workload's Pod template arrives whole, but only its requests, limits and +scheduling fields are kept: no environment value, command or image. Enter on +a row opens the object it is about. Nothing is changed: korvid does not +scale, retry or edit a policy for you. + ## Change scope without losing context korvid always watches exactly one explicit scope. The startup namespace diff --git a/src/korvid/__main__.py b/src/korvid/__main__.py index 57550bdbb..dca8b0d79 100644 --- a/src/korvid/__main__.py +++ b/src/korvid/__main__.py @@ -140,6 +140,7 @@ from korvid.ui.command_router import CommandRouter from korvid.ui.config_propagation_controller import ConfigPropagationController from korvid.ui.context_switch_coordinator import ContextSwitchCoordinator, ContextSwitchResult +from korvid.ui.creation_constraints_controller import CreationConstraintsController from korvid.ui.debug import DebugController, DebugSettings from korvid.ui.deployment_outcome_actions import DeploymentOutcomePodActions from korvid.ui.deployment_outcome_controller import DeploymentOutcomeController @@ -992,6 +993,15 @@ def _construct_app_runtime(app: KorvidApp, inputs: AppRuntimeInputs) -> AppRunti "pods", namespace, name, epoch=epoch, expected_uid=uid ), ) + creation_constraints = CreationConstraintsController( + ui=AppUiSurface(app), + view=view, + context=context, + client=inputs.evidence_client, + navigate=lambda kind, namespace, name, epoch, uid: workspace_ref.get().jump_to_object( + kind, namespace, name, epoch=epoch, expected_uid=uid + ), + ) bridge_dispatch = AppContextDispatch() inspect_surface = AppInspectSurface(app) inspect_controller = ResourceInspectController( @@ -1276,6 +1286,7 @@ def _construct_app_runtime(app: KorvidApp, inputs: AppRuntimeInputs) -> AppRunti slots=slots.unavailable_reason, outcomes=deployment_outcomes.unavailable_reason, propagation=config_propagation.unavailable_reason, + creation=creation_constraints.unavailable_reason, ), ) commands = CommandRouter( @@ -1290,6 +1301,7 @@ def _construct_app_runtime(app: KorvidApp, inputs: AppRuntimeInputs) -> AppRunti slots=slots, outcomes=deployment_outcomes, propagation=config_propagation, + creation=creation_constraints, ) return AppRuntime( view=view, diff --git a/src/korvid/ui/action_policy.py b/src/korvid/ui/action_policy.py index 7d1c63abb..9aec4abe9 100644 --- a/src/korvid/ui/action_policy.py +++ b/src/korvid/ui/action_policy.py @@ -211,6 +211,7 @@ def compose_command_reasons( slots: Callable[[], UnavailableReason | None] | None = None, outcomes: Callable[[], UnavailableReason | None] | None = None, propagation: Callable[[], UnavailableReason | None] | None = None, + creation: Callable[[], UnavailableReason | None] | None = None, ) -> dict[str, Callable[[], UnavailableReason | None]]: """Build `ActionPolicy(reason_by_command=...)` from its owners. @@ -258,6 +259,8 @@ def compose_command_reasons( propagation: `ConfigPropagationController.unavailable_reason` - whether `:consumers` has a connection and a selected ConfigMap or Secret to explain. + creation: `CreationConstraintsController.unavailable_reason` - + whether `:blocked` has a connection and a selected workload. Returns: The canonical command text -> reason-resolver map. @@ -286,6 +289,7 @@ def model_reason() -> UnavailableReason | None: **({"slots": slots} if slots is not None else {}), **({"outcomes": outcomes} if outcomes is not None else {}), **({"consumers": propagation} if propagation is not None else {}), + **({"blocked": creation} if creation is not None else {}), } diff --git a/src/korvid/ui/command.py b/src/korvid/ui/command.py index 77c74885e..a50abaac5 100644 --- a/src/korvid/ui/command.py +++ b/src/korvid/ui/command.py @@ -125,6 +125,22 @@ def __post_init__(self) -> None: ("propagation", "configmap", "secret", "mount", "subpath", "restart"), ), ), + CommandDescriptor( + aliases=("blocked",), + help=( + ( + ":blocked", + "Pods not created for a workload: recorded failures, quotas, LimitRanges", + ), + ), + operation=BuiltinOperation.CREATION_FAILURES, + maximum_arguments=0, + palette=PaletteCommand( + "Explain workload creation failures", + "blocked", + ("quota", "limitrange", "failedcreate", "admission", "pending", "replicas"), + ), + ), CommandDescriptor( aliases=("q", "quit"), help=((":q", "Quit (also :quit)"),), diff --git a/src/korvid/ui/command_router.py b/src/korvid/ui/command_router.py index 657efb3b9..4c39846d8 100644 --- a/src/korvid/ui/command_router.py +++ b/src/korvid/ui/command_router.py @@ -98,6 +98,12 @@ class PropagationCommands(Protocol): def open_selected(self) -> None: ... +class CreationCommands(Protocol): + """The workload creation constraints view's owner.""" + + def open_selected(self) -> None: ... + + class CommandRouter: """Dispatches typed commands to the owner that implements them.""" @@ -115,6 +121,7 @@ def __init__( slots: SlotCommands, outcomes: OutcomeCommands | None = None, propagation: PropagationCommands | None = None, + creation: CreationCommands | None = None, ) -> None: self._ui = ui self._agent = agent @@ -127,6 +134,7 @@ def __init__( self._slots = slots self._outcomes = outcomes self._propagation = propagation + self._creation = creation def route_builtin(self, command: BuiltinCommand) -> None: """Dispatch an app-owned command by canonical operation identity.""" @@ -140,6 +148,7 @@ def route_builtin(self, command: BuiltinCommand) -> None: BuiltinOperation.NAMESPACE_SLOTS: self._slots.open_reallocation, BuiltinOperation.DEPLOYMENT_OUTCOMES: self._open_outcomes, BuiltinOperation.CONFIG_PROPAGATION: self._open_propagation, + BuiltinOperation.CREATION_FAILURES: self._open_creation, } handler = no_argument_handlers.get(operation) if handler is not None: @@ -173,6 +182,14 @@ def _open_propagation(self) -> None: return self._propagation.open_selected() + def _open_creation(self) -> None: + if self._creation is None: + self._ui.notify( + "Workload creation failures are unavailable in this session", severity="warning" + ) + return + self._creation.open_selected() + def _route_agent(self, command: BuiltinCommand) -> None: if not self._agent.available: self._report_unknown(_builtin_text(command)) diff --git a/src/korvid/ui/context_switch_coordinator.py b/src/korvid/ui/context_switch_coordinator.py index 4d7f03347..6fcd98e05 100644 --- a/src/korvid/ui/context_switch_coordinator.py +++ b/src/korvid/ui/context_switch_coordinator.py @@ -64,6 +64,7 @@ UnavailableReason, ) from korvid.ui.config_propagation_controller import CONFIG_PROPAGATION_GROUP +from korvid.ui.creation_constraints_controller import CREATION_CONSTRAINTS_GROUP from korvid.ui.ui_surface import UiSurface from korvid.ui.view_state import ViewState from korvid.ui.widgets.pick_screen import PickScreen @@ -839,9 +840,10 @@ async def _teardown(self) -> None: # caller) so those pollers and workers release the old connection # before the wholesale watch stop below. await self._workspace().quiesce_for_context_switch() - # A configuration propagation run reads through the old client; its - # run token would drop the result anyway, but the reads must stop too. + # A diagnostic run reads through the old client; its run token would + # drop the result anyway, but the reads must stop too. self._surface.cancel_worker_group(CONFIG_PROPAGATION_GROUP) + self._surface.cancel_worker_group(CREATION_CONSTRAINTS_GROUP) # An old-cluster namespace prefetch still in flight could land after # the new cluster's and overwrite its completions — cancel it first. # Both steps run on the workspace controller: it owns the prefetch diff --git a/src/korvid/ui/creation_constraints_controller.py b/src/korvid/ui/creation_constraints_controller.py new file mode 100644 index 000000000..02cab92cd --- /dev/null +++ b/src/korvid/ui/creation_constraints_controller.py @@ -0,0 +1,164 @@ +"""Open the workload creation constraints view for the selected workload (#411). + +One exclusive worker reads the workload, its children, the Events and the +namespace's quotas and LimitRanges through `explain_creation`, under a run +token. The view opens only while that token is current: a context switch, +new credentials or a newer run drops the result, as `RunTracker` decides. So +does another dialog that is open when the reads finish, or a change of view, +scope or selection since they started. The context-switch coordinator +cancels the worker group before it swaps the client. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime + +from korvid.core.api_versions import observe_discovery +from korvid.core.capability import RunToken +from korvid.core.creation_constraints import CreationTarget, explain_creation +from korvid.core.creation_workloads import WORKLOAD_KINDS +from korvid.core.run_tokens import RunTracker +from korvid.k8s.discovery import canonical_resource_alias, resolve_resource +from korvid.k8s.evidence_requests import EvidenceClient +from korvid.ui.action_availability import AvailabilityCode, UnavailableReason +from korvid.ui.ui_surface import UiSurface +from korvid.ui.view_state import ViewState +from korvid.ui.widgets.creation_constraints_screen import CreationConstraintsScreen, Destination +from korvid.ui.workspace_controller import ContextGuard + +CREATION_CONSTRAINTS_GROUP = "creation-constraints" +_UNCONNECTED = UnavailableReason( + AvailabilityCode.MISSING_CAPABILITY, "Workload creation failures need a cluster connection" +) +_WRONG_VIEW = UnavailableReason( + AvailabilityCode.WRONG_VIEW, + "Open Deployments, ReplicaSets, StatefulSets, DaemonSets, Jobs or CronJobs to explain " + "creation failures", +) +_NO_SELECTION = UnavailableReason(AvailabilityCode.NO_SELECTION, "Select a workload") + +#: Opens an object in its table: (view alias, namespace, name, epoch, expected UID). +ObjectNavigator = Callable[[str, str, str, int, str | None], Awaitable[None]] +#: The view, scope and (namespace, name) selection a run was started from. +_Origin = tuple[str, str, tuple[str | None, str | None]] + + +def _now() -> datetime: + return datetime.now(UTC) + + +class CreationConstraintsController: + """Runs the creation diagnostic and shows its result while it is current.""" + + def __init__( + self, + *, + ui: UiSurface, + view: ViewState, + context: ContextGuard, + client: EvidenceClient | None, + navigate: ObjectNavigator, + clock: Callable[[], datetime] = _now, + ) -> None: + self._ui = ui + self._view = view + self._context = context + self._client = client + self._navigate = navigate + self._clock = clock + self._tracker = RunTracker( + epoch=context.epoch, + session_generation=lambda: client.session_generation if client is not None else 0, + switching=context.switching, + ) + + def _selected_kind(self) -> str | None: + meta = self._view.aliases().get(self._view.canonical_kind(self._view.current_kind())) + workload = WORKLOAD_KINDS.get(meta.kind) if meta is not None else None + if meta is None or workload is None or meta.group != workload.group: + return None + return meta.kind + + def unavailable_reason(self) -> UnavailableReason | None: + """Why the view cannot open now, or None. + + Only for a missing connection, the wrong view or no selection: a + source the cluster refuses never disables it, the view says so. + """ + if self._client is None: + return _UNCONNECTED + if self._selected_kind() is None: + return _WRONG_VIEW + namespace, name = self._view.selected_ns_name(notify=False) + if not namespace or not name: + return _NO_SELECTION + return None + + def open_selected(self) -> None: + """Explain the selected workload's creation failures in a worker.""" + reason = self.unavailable_reason() + if reason is not None: + self._ui.notify(reason.message, severity="warning", markup=False) + return + kind = self._selected_kind() + namespace, name = self._view.selected_ns_name() + if not self._context.reads_allowed() or kind is None or not namespace or not name: + return + target = CreationTarget(kind, namespace, name, self._view.selected_uid(namespace, name)) + token = self._tracker.next_run() + self._ui.run_worker( + self._explain(target, token, self._context.epoch(), self._origin()), + exclusive=True, + group=CREATION_CONSTRAINTS_GROUP, + exit_on_error=False, + ) + + def _origin(self) -> _Origin: + """The view, scope and selected row a run explains.""" + return ( + self._view.current_kind(), + self._view.current_scope(), + self._view.selected_ns_name(notify=False), + ) + + async def _explain( + self, target: CreationTarget, token: RunToken, epoch: int, origin: _Origin + ) -> None: + client = self._client + if client is None: + return + snapshot = observe_discovery(await client.discovery_snapshot()) + report = await explain_creation(client, snapshot, target, token=token, clock=self._clock) + # Another dialog is open, or the user has moved on, as the reads finish. + if self._ui.screen_depth() > 1 or self._origin() != origin: + return + screen = CreationConstraintsScreen(report, self._clock) + self._tracker.apply_if_current( + token, + lambda: self._ui.push_screen(screen, lambda chosen: self._open(chosen, epoch)), + ) + + def _open(self, destination: Destination | None, epoch: int) -> None: + """Open a row's object by its API identity, never by a colliding display alias.""" + if destination is None: + return + aliases = self._view.aliases() + meta = resolve_resource(aliases, destination.group, destination.resource) + if meta is None: + name = destination.resource + qualified = f"{name}.{destination.group}" if destination.group else name + self._ui.notify( + f"{qualified} is not discovered in this session", severity="warning", markup=False + ) + return + alias = canonical_resource_alias(aliases, meta) + self._ui.run_worker( + self._navigate(alias, destination.namespace, destination.name, epoch, destination.uid), + group=CREATION_CONSTRAINTS_GROUP, + exit_on_error=False, + ) + + async def cancel_workers(self) -> None: + """Stop a run holding the old client, before a context switch swaps it.""" + await self._ui.cancel_workers(CREATION_CONSTRAINTS_GROUP) diff --git a/src/korvid/ui/evidence_text.py b/src/korvid/ui/evidence_text.py index 8d2ab6961..3755255f4 100644 --- a/src/korvid/ui/evidence_text.py +++ b/src/korvid/ui/evidence_text.py @@ -71,17 +71,22 @@ def basis_sentence(basis: Basis) -> str: return BASIS_SENTENCES[basis] +def elapsed_text(at: datetime, now: datetime) -> str: + """How long ago `at` was, in its largest whole unit, such as `2m ago`.""" + seconds = (now - at).total_seconds() + if seconds < 0: + return "in the future" + for size, unit in _UNITS: + if seconds >= size: + return f"{int(seconds // size)}{unit} ago" + return "just now" + + def age_text(observed_at: datetime | None, now: datetime) -> str: """How long ago an observation was made, in its largest whole unit.""" if observed_at is None: return "not observed" - seconds = (now - observed_at).total_seconds() - if seconds < 0: - return "observed in the future" - for size, unit in _UNITS: - if seconds >= size: - return f"observed {int(seconds // size)}{unit} ago" - return "observed just now" + return f"observed {elapsed_text(observed_at, now)}" def _scope_text(scope: SourceScope) -> str: diff --git a/src/korvid/ui/messages.py b/src/korvid/ui/messages.py index 093d781d7..ef22db0de 100644 --- a/src/korvid/ui/messages.py +++ b/src/korvid/ui/messages.py @@ -121,6 +121,7 @@ class BuiltinOperation(Enum): NAMESPACE_SLOTS = "slots" DEPLOYMENT_OUTCOMES = "outcomes" CONFIG_PROPAGATION = "consumers" + CREATION_FAILURES = "blocked" class BuiltinCommand(Message): diff --git a/src/korvid/ui/widgets/creation_constraints_screen.py b/src/korvid/ui/widgets/creation_constraints_screen.py new file mode 100644 index 000000000..74c488020 --- /dev/null +++ b/src/korvid/ui/widgets/creation_constraints_screen.py @@ -0,0 +1,458 @@ +"""Why a workload's Pods are not being created: recorded failures and namespace policies (#411). + +`CreationConstraintsScreen` shows one `CreationReport`: the workload and its +counters, a line per evidence source from the shared renderer, then the +recorded creation failures, the namespace's ResourceQuotas and its +LimitRanges, and what the evidence does not establish. It performs no I/O, +but re-renders the evidence lines and notes against its clock while it is +open, so a policy list that ages past its freshness policy stops ruling +anything out. Enter on a row dismisses with the `Destination` it leads to; +Escape dismisses with `None`. + +Every cell is a plain `Text`, so cluster-provided names and messages are +shown literally. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from typing import ClassVar + +from rich.text import Text +from textual.app import ComposeResult +from textual.binding import Binding +from textual.screen import ModalScreen +from textual.widgets import DataTable, Footer, Static +from textual.widgets.data_table import ColumnKey + +from korvid.core.capability import Basis, EvidenceState, Observation, conclusive, is_fresh +from korvid.core.creation_constraints import CreationReport, declarations +from korvid.core.creation_failures import Failure, FailureSource, RejectionKind +from korvid.core.creation_policies import LimitRangeProjection, LimitValue, QuotaProjection +from korvid.core.creation_quantities import Usage +from korvid.core.creation_workloads import WORKLOAD_KINDS +from korvid.ui.evidence_text import ( + basis_sentence, + elapsed_text, + not_established, + observation_line, +) + +_COLUMNS = ("Object", "What", "Value", "Note") +_UNREADABLE = ("", "constraints could not be read", "", "") +#: The kinds whose own status can record a creation failure, as a ReplicaFailure condition. +_CONDITION_KINDS = frozenset({"Deployment", "ReplicaSet"}) +_FAILURES = "Recorded creation failures — what the controller recorded, when it was recorded" +_QUOTAS = "ResourceQuotas — current values; not a reservation, and the next create may still differ" +_LIMIT_RANGES = "LimitRanges — current constraints; other admission plugins can also change a Pod" +_QUOTA_KINDS = frozenset( + { + RejectionKind.QUOTA_EXCEEDED, + RejectionKind.QUOTA_REQUIRES_FIELD, + RejectionKind.QUOTA_STATUS_UNKNOWN, + } +) +_LIMIT_KINDS = frozenset( + {RejectionKind.LIMIT_MINIMUM, RejectionKind.LIMIT_MAXIMUM, RejectionKind.LIMIT_RATIO} +) +#: Each LimitRange constraint and its field name, in display order. +_LIMIT_FIELDS: tuple[tuple[str, Callable[[LimitValue], str | None]], ...] = ( + ("default", lambda value: value.default), + ("defaultRequest", lambda value: value.default_request), + ("min", lambda value: value.min), + ("max", lambda value: value.max), + ("maxLimitRequestRatio", lambda value: value.max_limit_request_ratio), +) + + +@dataclass(frozen=True, slots=True) +class Destination: + """The object a row leads to: its API group and resource, namespace, name and UID. + + The group is kept so the caller resolves the resource by identity, not by a + display alias another API may have claimed. + """ + + group: str + resource: str + namespace: str + name: str + uid: str | None = None + + +@dataclass(frozen=True, slots=True) +class Heading: + """A section heading: what the section shows, then how to read it.""" + + text: str + + +@dataclass(frozen=True, slots=True) +class Row: + """One row's four cells, and where Enter on it leads.""" + + cells: tuple[str, str, str, str] + destination: Destination | None = None + + +TableRow = Heading | Row + + +def title_text(report: CreationReport) -> str: + """The workload, and the counters its controller last reported.""" + target = report.target + title = f"{target.kind} {target.namespace}/{target.name}" + if report.workload is None: + return title + counters = " · ".join(f"{c.label} {c.value or '?'}" for c in report.workload.counters) + return f"{title} {counters}" + + +def _observations(report: CreationReport) -> dict[str, Observation | None]: + return { + "workload": report.workload_observation, + "children": report.children_observation, + "events": report.events_observation, + "quotas": report.quotas_observation, + "limit-ranges": report.limit_ranges_observation, + } + + +def evidence_lines(report: CreationReport, now: datetime) -> list[Text]: + """Each source's line, in reading order, judged by its own policy.""" + observations = _observations(report) + lines: list[Text] = [] + for declaration in declarations(report.target): + observation = observations[declaration.key] + if observation is not None: + lines.append(observation_line(observation, declaration.policy, now)) + return lines + + +def _quota_text(name: str, report: CreationReport, now: datetime) -> str: + if any(quota.name == name for quota in report.quotas): + return f"quota {name}" + if report.quotas_conclusive(now): + return f"quota {name} (no longer present)" + return f"quota {name} (not in the quotas read)" + + +def _rejection_text(failure: Failure, report: CreationReport, now: datetime) -> str: + rejection = failure.rejection + kinds = "; ".join(kind.value for kind in rejection.kinds) + targets = [_quota_text(name, report, now) for name in rejection.quotas] + if any(kind in _LIMIT_KINDS for kind in rejection.kinds): + targets.append("a LimitRange in this namespace") + return f"{', '.join(targets)} — {kinds}" if targets else kinds + + +def _failure_row(failure: Failure, report: CreationReport, now: datetime) -> Row: + subject = failure.subject + label = f"{subject.kind} {subject.name}" + if failure.source is FailureSource.CONDITION: + label = f"{label} (condition)" + when = "undated" if failure.at is None else f"last {elapsed_text(failure.at, now)}" + workload = WORKLOAD_KINDS[subject.kind] + destination = Destination( + workload.group, workload.resource, report.target.namespace, subject.name, subject.uid + ) + times = "once" if failure.count == 1 else f"{failure.count} times" + cells = (label, f"{times} · {when}", _rejection_text(failure, report, now)) + return Row((*cells, failure.rejection.message), destination) + + +def _scopes_text(quota: QuotaProjection) -> str: + parts = list(quota.scopes) + parts += [ + f"{e.scope_name} {e.operator}" + (f" {', '.join(e.values)}" if e.values else "") + for e in quota.scope_selector + ] + return "; ".join(parts) + + +def _marker(quota: QuotaProjection, usage: Usage) -> str: + if quota.stale: + return "status not yet updated" + if usage is Usage.NOT_ESTABLISHED: + return "not established" + return "at its limit" if usage is Usage.AT_LIMIT else "" + + +def _quota_rows(quota: QuotaProjection, namespace: str) -> list[Row]: + destination = Destination("", "resourcequotas", namespace, quota.name, quota.uid) + cells: list[tuple[str, str, str, str]] = [ + ( + "", + f"{entry.group.value}: {entry.resource}", + f"used {entry.used or '?'} / hard {entry.hard}", + _marker(quota, entry.usage), + ) + for entry in quota.entries + ] + if not quota.complete: + cells.append(_UNREADABLE) + elif not cells: + cells.append(("", "no hard limits", "", "")) + scopes = _scopes_text(quota) + if scopes: + cells.append(("", "scopes", scopes, "")) + cells[0] = (quota.name, *cells[0][1:]) + return [Row(row, destination) for row in cells] + + +def _limit_text(value: LimitValue) -> str: + return " · ".join( + f"{name} {quantity}" + for name, read in _LIMIT_FIELDS + if (quantity := read(value)) is not None + ) + + +def _limit_range_rows(limit_range: LimitRangeProjection, namespace: str) -> list[Row]: + destination = Destination("", "limitranges", namespace, limit_range.name, limit_range.uid) + cells: list[tuple[str, str, str, str]] = [ + ("", f"{value.type} {value.resource}", _limit_text(value), "") + for value in limit_range.values + ] + if not limit_range.complete: + cells.append(_UNREADABLE) + elif not cells: + cells.append(("", "no constraints", "", "")) + cells[0] = (limit_range.name, *cells[0][1:]) + return [Row(row, destination) for row in cells] + + +def table_rows(report: CreationReport, now: datetime) -> list[TableRow]: + """The failures, quotas and LimitRanges under their headings; an empty section is left out.""" + namespace = report.target.namespace + sections: list[tuple[str, list[Row]]] = [ + (_FAILURES, [_failure_row(f, report, now) for f in report.failures]), + (_QUOTAS, [row for q in report.quotas for row in _quota_rows(q, namespace)]), + ( + _LIMIT_RANGES, + [row for r in report.limit_ranges for row in _limit_range_rows(r, namespace)], + ), + ] + rows: list[TableRow] = [] + for heading, section in sections: + if section: + rows.append(Heading(heading)) + rows += section + return rows + + +def _workload_notes(report: CreationReport, now: datetime) -> list[Text]: + kind = report.target.kind + observation = report.workload_observation + if observation.basis is Basis.OBJECT_ABSENT: + if not is_fresh(observation, declarations(report.target)[0].policy, now): + return [ + Text(f"The {kind} did not exist when it was read; it may have been created since.") + ] + return [Text(f"The {kind} no longer exists.")] + notes: list[Text] = [] + if report.recreated: + notes.append( + Text( + f"The {kind} was recreated since it was selected: only the new one's " + "evidence is shown." + ) + ) + if report.workload is not None: + notes.append( + Text("The counters are what the controller last reported, not a count of Pods.") + ) + return notes + + +def _failure_notes(report: CreationReport) -> list[Text]: + target = report.target + notes: list[Text] = [] + if report.earlier_failures: + children = f" or its {target.child}s" if target.child is not None else "" + notes.append( + Text( + f"{report.earlier_failures} creation failures recorded about earlier " + f"incarnations of this {target.kind}{children} are not shown." + ) + ) + if report.failures: + notes.append( + Text( + "The rejections were recorded at the times shown; the quotas and LimitRanges " + "are as read now." + ) + ) + elif report.events_observation.state is EvidenceState.AVAILABLE: + notes += _nothing_recorded(report) + else: + reason = basis_sentence(report.events_observation.basis) + notes.append(not_established(f"whether a creation failure is recorded: {reason}")) + return notes + + +def _nothing_recorded(report: CreationReport) -> list[Text]: + """No failure in what was read: never a conclusion, and only about what was read.""" + target = report.target + children = report.children_observation + subjects_read = report.workload_observation.state is EvidenceState.AVAILABLE and ( + children is None or children.state is EvidenceState.AVAILABLE + ) + sources = ( + "Events and conditions" if subjects_read and target.kind in _CONDITION_KINDS else "Events" + ) + notes = [ + Text( + f"No creation failure is recorded in the {sources} read. Events expire, so this " + "does not show that none happened." + ) + ] + if not subjects_read: + about = target.kind + (f" or its {target.child}s" if target.child is not None else "") + notes.append( + not_established( + f"whether a failure is recorded about the {about}: they could not all be read" + ) + ) + return notes + + +def _listed( + report: CreationReport, key: str, observation: Observation | None, now: datetime +) -> bool | str: + """`True` when the list is conclusive, otherwise why it is not.""" + if observation is None: + return True + policy = next(d.policy for d in declarations(report.target) if d.key == key) + if conclusive(observation, policy, now): + return True + if observation.state is EvidenceState.AVAILABLE and not is_fresh(observation, policy, now): + return "the list is stale" + return basis_sentence(observation.basis) + + +def _completeness_notes(report: CreationReport, now: datetime) -> list[Text]: + namespace = report.target.namespace + notes: list[Text] = [] + child = report.target.child + children = _listed(report, "children", report.children_observation, now) + if child is not None and children is not True: + notes.append(not_established(f"whether every {child} is listed: {children}")) + for key, observation, items, kind in ( + ("quotas", report.quotas_observation, report.quotas, "ResourceQuota"), + ("limit-ranges", report.limit_ranges_observation, report.limit_ranges, "LimitRange"), + ): + listed = _listed(report, key, observation, now) + if listed is not True: + notes.append(not_established(f"whether every {kind} is listed: {listed}")) + elif not items: + notes.append(Text(f"No {kind} in namespace {namespace}.")) + return notes + + +def report_notes(report: CreationReport, now: datetime) -> list[Text]: + """What the evidence says about the workload, and what it does not establish.""" + return [ + *_workload_notes(report, now), + *_failure_notes(report), + *_completeness_notes(report, now), + ] + + +class CreationConstraintsScreen(ModalScreen[Destination | None]): + """One workload's recorded creation failures and its namespace's policies.""" + + BINDINGS: ClassVar[list[Binding | tuple[str, str] | tuple[str, str, str]]] = [ + Binding("escape", "close", "Close", show=True), + Binding("q", "close", "Close", show=False), + ] + + DEFAULT_CSS = """ + CreationConstraintsScreen { + layout: vertical; + background: $background; + } + CreationConstraintsScreen #creation-title { + padding: 0 1; + text-style: bold; + } + CreationConstraintsScreen #creation-evidence, CreationConstraintsScreen #creation-notes { + padding: 0 1; + color: $text-muted; + } + CreationConstraintsScreen DataTable { + height: 1fr; + } + """ + + def __init__( + self, + report: CreationReport, + clock: Callable[[], datetime], + *, + refresh_seconds: float = 5.0, + ) -> None: + super().__init__() + self.report = report + self._clock = clock + self._refresh_seconds = refresh_seconds + self._rows: dict[str, Destination] = {} + self._columns: list[ColumnKey] = [] + + def compose(self) -> ComposeResult: + yield Footer() + yield Static(title_text(self.report), id="creation-title", markup=False) + yield Static(id="creation-evidence") + yield DataTable[Text](id="creation-table") + yield Static(id="creation-notes") + + def _render_text(self) -> None: + now = self._clock() + evidence = Text("\n").join(evidence_lines(self.report, now)) + self.query_one("#creation-evidence", Static).update(evidence) + notes = Text("\n").join(report_notes(self.report, now)) + self.query_one("#creation-notes", Static).update(notes) + self._refresh_rows(now) + + def _refresh_rows(self, now: datetime) -> None: + """Re-render cells that depend on the time: ages, and whether a list still rules out.""" + if not self._columns: + return + table: DataTable[Text] = self.query_one(DataTable) + for index, row in enumerate(table_rows(self.report, now)): + if isinstance(row, Row): + for column, cell in zip(self._columns, row.cells, strict=True): + table.update_cell(f"r{index}", column, Text(cell)) + + def on_mount(self) -> None: + self._render_text() + self.set_interval(self._refresh_seconds, self._render_text) + table: DataTable[Text] = self.query_one(DataTable) + table.cursor_type = "row" + self._columns = table.add_columns(*_COLUMNS) + first: int | None = None + for index, row in enumerate(table_rows(self.report, self._clock())): + if isinstance(row, Heading): + title, _, meaning = row.text.partition(" — ") + table.add_row(Text(title, style="bold"), Text(""), Text(meaning), key=f"h{index}") + continue + key = f"r{index}" + if row.destination is not None: + self._rows[key] = row.destination + # A rejection message can span lines; the row grows to show all of it. + table.add_row(*(Text(cell) for cell in row.cells), key=key, height=None) + first = index if first is None else first + if first is not None: + table.move_cursor(row=first) + table.focus() + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + """Enter on a row returns where it leads; a heading does nothing.""" + destination = self._rows.get(str(event.row_key.value)) + if destination is not None: + self.dismiss(destination) + + def action_close(self) -> None: + self.dismiss(None) diff --git a/src/korvid/ui/worker_failures.py b/src/korvid/ui/worker_failures.py index 8f0db2f65..7d44cca86 100644 --- a/src/korvid/ui/worker_failures.py +++ b/src/korvid/ui/worker_failures.py @@ -12,6 +12,7 @@ from types import MappingProxyType from korvid.ui.config_propagation_controller import CONFIG_PROPAGATION_GROUP +from korvid.ui.creation_constraints_controller import CREATION_CONSTRAINTS_GROUP from korvid.ui.session_timeline_controller import TIMELINE_EVENT_GROUP, TIMELINE_NAVIGATION_GROUP from korvid.ui.workspace_controller import RELATIONSHIP_GROUP @@ -22,5 +23,6 @@ RELATIONSHIP_GROUP: "Relationships failed", TIMELINE_NAVIGATION_GROUP: "Timeline navigation failed", CONFIG_PROPAGATION_GROUP: "Configuration propagation failed", + CREATION_CONSTRAINTS_GROUP: "Workload creation failure explanation failed", } ) diff --git a/tests/ui/test_action_policy.py b/tests/ui/test_action_policy.py index c96567485..ea446ea20 100644 --- a/tests/ui/test_action_policy.py +++ b/tests/ui/test_action_policy.py @@ -847,3 +847,25 @@ def test_every_search_key_has_an_owner() -> None: """The same audit for `n`/`N`: `SEARCH_ACTIONS` is the pane-search vocabulary, and the inspect owner is who answers for it.""" assert set(SEARCH_ACTIONS) <= set(_INSPECT_REASON_ACTIONS) + + +def test_composed_command_reasons_ask_the_creation_owner_about_blocked() -> None: + """`:blocked` is greyed out by its owner's own answer: no connection, the + wrong view, or no selected workload.""" + wrong_view = UnavailableReason(AvailabilityCode.WRONG_VIEW, "Open a workload view") + answer: UnavailableReason | None = wrong_view + reasons = compose_command_reasons( + agent_available=lambda: True, + agent_setup_available=lambda: True, + agent_session_configured=lambda: True, + mcp=lambda: None, + telepresence=lambda: None, + proposals=lambda: None, + namespace=lambda: None, + context=lambda: None, + port_forwards=lambda: None, + creation=lambda: answer, + ) + assert reasons["blocked"]() == wrong_view + answer = None + assert reasons["blocked"]() is None diff --git a/tests/ui/test_command_router.py b/tests/ui/test_command_router.py index 739aba013..e37ddc861 100644 --- a/tests/ui/test_command_router.py +++ b/tests/ui/test_command_router.py @@ -114,6 +114,7 @@ def __init__( catalog_missing: bool = False, outcomes: FakeOutcomes | None = None, propagation: FakePropagation | None = None, + creation: FakePropagation | None = None, ) -> None: self.ui = FakeUi() self.agent = FakeAgent(agent_available) @@ -136,6 +137,7 @@ def __init__( slots=self.slots, outcomes=outcomes, propagation=propagation, + creation=creation, ) @@ -297,3 +299,23 @@ def test_configuration_propagation_without_owner_warns_unavailable() -> None: assert harness.ui.notifications == [ ("Configuration propagation is unavailable in this session", "warning") ] + + +def test_creation_failures_reach_their_owner_once() -> None: + creation = FakePropagation() + harness = Harness(creation=creation) + + harness.router.route_builtin(_builtin(BuiltinOperation.CREATION_FAILURES)) + + assert creation.opens == 1 + assert harness.ui.notifications == [] + + +def test_creation_failures_without_owner_warn_unavailable() -> None: + harness = Harness() + + harness.router.route_builtin(_builtin(BuiltinOperation.CREATION_FAILURES)) + + assert harness.ui.notifications == [ + ("Workload creation failures are unavailable in this session", "warning") + ] diff --git a/tests/ui/test_context_switch_coordinator.py b/tests/ui/test_context_switch_coordinator.py index 4cd4fcfe4..72049bd77 100644 --- a/tests/ui/test_context_switch_coordinator.py +++ b/tests/ui/test_context_switch_coordinator.py @@ -39,6 +39,7 @@ SessionConfiguration, SwitchAgent, ) +from korvid.ui.creation_constraints_controller import CREATION_CONSTRAINTS_GROUP from korvid.ui.session_timeline_controller import SessionTimelineController from korvid.ui.widgets.pick_screen import PickScreen @@ -481,6 +482,7 @@ def _forward_record(port: int) -> ForwardRecord: "describe-hide", "workspace-quiesce", f"workers-cancelled:{CONFIG_PROPAGATION_GROUP}", + f"workers-cancelled:{CREATION_CONSTRAINTS_GROUP}", "ns-prefetch-cancelled", "ns-words-cleared", "watch-stop-all", diff --git a/tests/ui/test_creation_constraints_controller.py b/tests/ui/test_creation_constraints_controller.py new file mode 100644 index 000000000..8c037bee3 --- /dev/null +++ b/tests/ui/test_creation_constraints_controller.py @@ -0,0 +1,377 @@ +"""The creation constraints controller: availability, runs and late results (#411).""" + +from __future__ import annotations + +from collections.abc import Awaitable, Mapping +from datetime import UTC, datetime +from typing import Any + +from korvid.core.creation_constraints import CreationTarget +from korvid.core.creation_failures import Subject +from korvid.k8s.discovery import ResourceMeta +from korvid.k8s.discovery_snapshot import ( + GroupVersionOutcome, + RawDiscovery, + RootOutcome, + ServedResource, +) +from korvid.k8s.evidence_requests import EvidenceClient, EvidencePage, EvidenceTarget +from korvid.ui.action_availability import AvailabilityCode +from korvid.ui.creation_constraints_controller import ( + CREATION_CONSTRAINTS_GROUP, + CreationConstraintsController, +) +from korvid.ui.view_state import ViewState +from korvid.ui.widgets.creation_constraints_screen import CreationConstraintsScreen, Destination +from korvid.ui.workspace_controller import ContextGuard + +from .test_session_timeline_controller import FakeUiSurface + +NOW = datetime(2026, 10, 30, 10, 0, tzinfo=UTC) +DEPLOYMENTS = ResourceMeta("Deployment", "deployments", "apps", "v1", True) +CRON_JOBS = ResourceMeta("CronJob", "cronjobs", "batch", "v1", True) +PODS = ResourceMeta("Pod", "pods", "", "v1", True) +QUOTAS = ResourceMeta("ResourceQuota", "resourcequotas", "", "v1", True) +REPLICA_SETS = ResourceMeta("ReplicaSet", "replicasets", "apps", "v1", True) +FOREIGN_REPLICA_SETS = ResourceMeta("ReplicaSet", "replicasets", "vendor.example.com", "v1", True) +FOREIGN = ResourceMeta("Deployment", "deployments", "vendor.example.com", "v1", True) +_SERVED = { + ("", "v1"): ("resourcequotas", "limitranges", "events"), + ("apps", "v1"): ("deployments", "replicasets"), + ("batch", "v1"): ("jobs", "cronjobs"), +} +EVENT = { + "metadata": {"name": "e", "namespace": "shop", "uid": "uid-e"}, + "reason": "FailedCreate", + "involvedObject": {"kind": "ReplicaSet", "name": "api-7d9f", "uid": "uid-rs"}, + "message": 'pods "x" is forbidden: exceeded quota: compute, requested: pods=1', +} + + +class FakeView(ViewState): + def __init__(self, kind: str = "deployments", selected: tuple[str, str] | None = None) -> None: + self.kind = kind + self.scope = "shop" + self.selected = selected + self.alias_map = { + "deployments": DEPLOYMENTS, + "cronjobs": CRON_JOBS, + "pods": PODS, + "deployments.vendor.example.com": FOREIGN, + "resourcequotas": QUOTAS, + } + + def current_kind(self) -> str: + return self.kind + + def canonical_kind(self, kind: str) -> str: + return kind + + def aliases(self) -> Mapping[str, ResourceMeta]: + return self.alias_map + + def selected_ns_name(self, *, notify: bool = True) -> tuple[str | None, str | None]: + return self.selected if self.selected is not None else (None, None) + + def selected_uid(self, namespace: str | None, name: str) -> str | None: + return f"uid-{name}" + + def current_scope(self) -> str: + return self.scope + + def resources(self, kind: str, scope: str) -> list[Any]: + return [] + + def readonly(self) -> bool: + return False + + def default_namespace(self) -> str | None: + return "shop" + + def gvr_label(self, meta: ResourceMeta) -> str: + return meta.plural + + def write_locus(self, namespace: str | None) -> str: + return namespace or "" + + +class FakeContext(ContextGuard): + def __init__(self) -> None: + self.current = 1 + self.in_switch = False + self.allowed = True + + def epoch(self) -> int: + return self.current + + def switching(self) -> bool: + return self.in_switch + + def reads_allowed(self) -> bool: + return self.allowed + + def crossed(self, epoch: int) -> bool: + return self.in_switch or epoch != self.current + + +class FakeClient(EvidenceClient): + def __init__(self) -> None: + self.generation = 3 + self.gets: list[tuple[EvidenceTarget, str]] = [] + self.lists: list[EvidenceTarget] = [] + + @property + def session_generation(self) -> int: + return self.generation + + async def discovery_snapshot(self) -> RawDiscovery: + outcomes = tuple( + GroupVersionOutcome( + group, version, tuple(ServedResource(r, frozenset({"get", "list"})) for r in names) + ) + for (group, version), names in _SERVED.items() + ) + groups = tuple((g, v) for g, v in _SERVED if g) + return RawDiscovery( + self.generation, + NOW, + 0.0, + RootOutcome("/api", (("", "v1"),)), + RootOutcome("/apis", groups), + outcomes, + ) + + async def list_evidence_page( + self, target: EvidenceTarget, *, limit: int, continuation: str, max_bytes: int + ) -> EvidencePage: + self.lists.append(target) + if target.resource == "replicasets": + owner = [{"uid": "uid-api", "controller": True}] + rs = { + "metadata": { + "name": "api-7d9f", + "namespace": "shop", + "uid": "uid-rs", + "ownerReferences": owner, + } + } + return EvidencePage((rs,)) + if target.resource == "events": + return EvidencePage((EVENT,)) + return EvidencePage(()) + + async def get_evidence_object( + self, target: EvidenceTarget, name: str, *, max_bytes: int + ) -> dict[str, Any]: + self.gets.append((target, name)) + return { + "metadata": {"name": name, "namespace": "shop", "uid": f"uid-{name}"}, + "spec": {}, + "status": {}, + } + + +class Harness: + def __init__(self, *, client: FakeClient | None = None, view: FakeView | None = None) -> None: + self.ui = FakeUiSurface(close_worker_coroutines=False) + self.view = view or FakeView(selected=("shop", "api")) + self.context = FakeContext() + self.client = FakeClient() if client is None else client + self.navigations: list[tuple[str, str, str, int, str | None]] = [] + self.controller = CreationConstraintsController( + ui=self.ui, + view=self.view, + context=self.context, + client=self.client, + navigate=self._navigate, + clock=lambda: NOW, + ) + + async def _navigate( + self, kind: str, namespace: str, name: str, epoch: int, uid: str | None + ) -> None: + self.navigations.append((kind, namespace, name, epoch, uid)) + + async def run(self, index: int = -1) -> None: + work = self.ui.workers[index] + assert isinstance(work, Awaitable) + await work + + +def _code(harness: Harness) -> AvailabilityCode | None: + reason = harness.controller.unavailable_reason() + return None if reason is None else reason.code + + +def test_it_is_available_on_a_selected_workload() -> None: + assert _code(Harness()) is None + assert _code(Harness(view=FakeView("cronjobs", ("shop", "nightly")))) is None + + +def test_it_says_why_it_is_unavailable() -> None: + unconnected = CreationConstraintsController( + ui=FakeUiSurface(), + view=FakeView(selected=("shop", "api")), + context=FakeContext(), + client=None, + navigate=Harness()._navigate, + ) + + reason = unconnected.unavailable_reason() + assert reason is not None + assert reason.code is AvailabilityCode.MISSING_CAPABILITY + assert _code(Harness(view=FakeView("pods", ("shop", "api-1")))) is AvailabilityCode.WRONG_VIEW + assert ( + _code(Harness(view=FakeView("deployments.vendor.example.com", ("shop", "x")))) + is AvailabilityCode.WRONG_VIEW + ) + assert _code(Harness(view=FakeView("deployments", None))) is AvailabilityCode.NO_SELECTION + + +async def test_opening_reads_the_selected_workload_and_shows_its_failures() -> None: + harness = Harness() + + harness.controller.open_selected() + await harness.run() + + assert harness.ui.worker_groups == [(CREATION_CONSTRAINTS_GROUP, False)] + assert harness.ui.exclusive_workers == [True] + (screen,) = harness.ui.screens + assert isinstance(screen, CreationConstraintsScreen) + assert screen.report.target == CreationTarget("Deployment", "shop", "api", "uid-api") + assert [f.subject for f in screen.report.failures] == [ + Subject("ReplicaSet", "api-7d9f", "uid-rs") + ] + assert harness.client.gets == [(EvidenceTarget("apps", "v1", "deployments", "shop"), "api")] + assert {t.resource for t in harness.client.lists} == { + "replicasets", + "events", + "resourcequotas", + "limitranges", + } + + +def test_opening_where_it_is_unavailable_says_why_and_reads_nothing() -> None: + harness = Harness(view=FakeView("pods", ("shop", "api-1"))) + + harness.controller.open_selected() + + assert harness.ui.workers == [] + assert [n.severity for n in harness.ui.notifications] == ["warning"] + + +def test_opening_while_reads_are_held_reads_nothing() -> None: + harness = Harness() + harness.context.allowed = False + + harness.controller.open_selected() + + assert harness.ui.workers == [] + + +async def test_a_newer_run_drops_an_older_result() -> None: + harness = Harness() + + harness.controller.open_selected() + harness.controller.open_selected() + await harness.run(0) + + assert harness.ui.screens == [] + await harness.run(1) + assert len(harness.ui.screens) == 1 + + +async def test_a_context_switch_or_new_credentials_drop_the_result() -> None: + switched = Harness() + reauthenticated = Harness() + + switched.controller.open_selected() + reauthenticated.controller.open_selected() + switched.context.current = 2 + reauthenticated.client.generation = 4 + await switched.run() + await reauthenticated.run() + + assert (switched.ui.screens, reauthenticated.ui.screens) == ([], []) + + +async def test_choosing_a_row_opens_its_object_in_the_run_epoch() -> None: + harness = Harness() + harness.controller.open_selected() + await harness.run() + + assert harness.ui.screen_callback is not None + harness.ui.screen_callback(None) + assert len(harness.ui.workers) == 1 + harness.context.current = 2 + harness.ui.screen_callback(Destination("", "resourcequotas", "shop", "compute", "uid-q")) + await harness.run() + + assert harness.navigations == [("resourcequotas", "shop", "compute", 1, "uid-q")] + assert harness.ui.worker_groups[-1] == (CREATION_CONSTRAINTS_GROUP, False) + + +async def test_a_dialog_opened_during_the_reads_keeps_the_view_closed() -> None: + harness = Harness() + harness.controller.open_selected() + harness.ui.screen_depth = lambda: 2 # type: ignore[method-assign] # a dialog opened + + await harness.run() + + assert harness.ui.screens == [] + + +async def test_leaving_the_view_or_the_selection_keeps_the_result_closed() -> None: + moved = Harness() + rescoped = Harness() + reselected = Harness() + + for harness in (moved, rescoped, reselected): + harness.controller.open_selected() + moved.view.kind = "pods" + rescoped.view.scope = "other" + reselected.view.selected = ("shop", "other") + for harness in (moved, rescoped, reselected): + await harness.run() + + assert (moved.ui.screens, rescoped.ui.screens, reselected.ui.screens) == ([], [], []) + + +async def test_the_worker_group_is_cancelled_on_request() -> None: + harness = Harness() + + await harness.controller.cancel_workers() + + assert harness.ui.cancelled_groups == [CREATION_CONSTRAINTS_GROUP] + + +async def _choose(harness: Harness, destination: Destination) -> None: + harness.controller.open_selected() + await harness.run() + assert harness.ui.screen_callback is not None + harness.ui.screen_callback(destination) + await harness.run() + + +async def test_a_destination_is_opened_under_its_own_api_group_not_a_colliding_alias() -> None: + harness = Harness() + harness.view.alias_map["replicasets"] = FOREIGN_REPLICA_SETS + harness.view.alias_map["replicasets.apps"] = REPLICA_SETS + + await _choose(harness, Destination("apps", "replicasets", "shop", "api-7d9f", "uid-rs")) + + assert harness.navigations == [("replicasets.apps", "shop", "api-7d9f", 1, "uid-rs")] + + +async def test_a_destination_this_session_does_not_serve_is_said_not_opened() -> None: + harness = Harness() + + harness.controller.open_selected() + await harness.run() + assert harness.ui.screen_callback is not None + harness.ui.screen_callback(Destination("apps", "replicasets", "shop", "api-7d9f", "uid-rs")) + + assert harness.navigations == [] + assert [(n.message, n.severity) for n in harness.ui.notifications] == [ + ("replicasets.apps is not discovered in this session", "warning") + ] diff --git a/tests/ui/test_creation_constraints_screen.py b/tests/ui/test_creation_constraints_screen.py new file mode 100644 index 000000000..8710bd40d --- /dev/null +++ b/tests/ui/test_creation_constraints_screen.py @@ -0,0 +1,562 @@ +"""The workload creation constraints view (#411): rows, notes and keys.""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from rich.text import Text +from textual.app import App +from textual.coordinate import Coordinate +from textual.widgets import DataTable, Static + +from korvid.core.capability import ( + ApiSource, + Baseline, + Basis, + EvidenceState, + Observation, + SourceScope, + Verb, +) +from korvid.core.creation_constraints import CreationReport, CreationTarget +from korvid.core.creation_failures import Failure, FailureSource, Subject, classify +from korvid.core.creation_policies import ( + LimitRangeProjection, + LimitValue, + QuotaEntry, + QuotaProjection, + ScopeExpression, +) +from korvid.core.creation_quantities import QuotaGroup, Usage +from korvid.core.creation_workloads import Counter, WorkloadProjection +from korvid.ui.widgets.creation_constraints_screen import ( + CreationConstraintsScreen, + Destination, + Heading, + Row, + evidence_lines, + report_notes, + table_rows, + title_text, +) +from tests.ui.waits import screen_ready, until + +NOW = datetime(2026, 10, 30, 10, 0, tzinfo=UTC) +TARGET = CreationTarget("Deployment", "shop", "api", "uid-api") +SCOPE = SourceScope.namespace("shop") +DEPLOYMENT = Subject("Deployment", "api", "uid-api") +REPLICA_SET = Subject("ReplicaSet", "api-7d9f", "uid-rs") +EXCEEDED = ( + 'Error creating: pods "api-7d9f-x" is forbidden: exceeded quota: compute, ' + "requested: pods=1, used: pods=20, limited: pods=20" +) + + +def _source(group: str, resource: str, verb: Verb = Verb.LIST, **extra: Any) -> ApiSource: + return ApiSource(group, resource, SCOPE, "v1", verb, **extra) + + +def _seen(source: ApiSource, basis: Basis = Basis.OBSERVED, uid: str | None = None) -> Observation: + state = { + Basis.OBSERVED: EvidenceState.AVAILABLE, + Basis.OBJECT_ABSENT: EvidenceState.AVAILABLE, + Basis.TRUNCATED: EvidenceState.PARTIAL, + Basis.FORBIDDEN: EvidenceState.DENIED, + Basis.STALE_STATUS: EvidenceState.PARTIAL, + }[basis] + return Observation(source, state, basis, NOW - timedelta(seconds=2), observed_uid=uid) + + +WORKLOAD_SOURCE = _source("apps", "deployments", Verb.GET, name="api", expected_uid="uid-api") +WORKLOAD = WorkloadProjection( + DEPLOYMENT, (Counter("desired", "3"), Counter("created", "0")), None, None +) +QUOTA = QuotaProjection( + "compute", + "uid-q", + ( + QuotaEntry("requests.cpu", QuotaGroup.COMPUTE, "2", "2", Usage.AT_LIMIT), + QuotaEntry("pods", QuotaGroup.COUNT, "20", "20", Usage.AT_LIMIT), + QuotaEntry("services", QuotaGroup.COUNT, "5", "1", Usage.BELOW), + ), + ("NotTerminating",), + (ScopeExpression("PriorityClass", "In", ("gold", "silver")),), +) +LIMITS = LimitRangeProjection( + "defaults", + "uid-lr", + (LimitValue("Container", "cpu", max="1", default="500m", default_request="100m"),), +) +FAILURE = Failure( + REPLICA_SET, FailureSource.EVENT, classify(EXCEEDED), 12, NOW - timedelta(minutes=2) +) + + +def _report(**changes: Any) -> CreationReport: + report = CreationReport( + target=TARGET, + workload_observation=_seen(WORKLOAD_SOURCE, uid="uid-api"), + children_observation=_seen(_source("apps", "replicasets")), + events_observation=_seen(_source("events.k8s.io", "events")), + quotas_observation=_seen(_source("", "resourcequotas")), + limit_ranges_observation=_seen(_source("", "limitranges")), + workload=WORKLOAD, + children=(), + failures=(FAILURE,), + earlier_failures=0, + quotas=(QUOTA,), + limit_ranges=(LIMITS,), + baseline=Baseline(EvidenceState.AVAILABLE), + ) + return replace(report, **changes) + + +def _plain(lines: list[Text]) -> list[str]: + return [line.plain for line in lines] + + +def _rows(report: CreationReport) -> list[Row]: + return [row for row in table_rows(report, NOW) if isinstance(row, Row)] + + +def test_the_title_names_the_workload_and_its_counters() -> None: + assert title_text(_report()) == "Deployment shop/api desired 3 · created 0" + assert title_text(_report(workload=None)) == "Deployment shop/api" + unset = replace(WORKLOAD, counters=(Counter("desired", None), Counter("created", "0"))) + assert title_text(_report(workload=unset)) == "Deployment shop/api desired ? · created 0" + + +def test_each_source_has_a_line_from_the_shared_renderer_in_order() -> None: + lines = _plain(evidence_lines(_report(), NOW)) + + assert [line.split(":")[0] for line in lines] == [ + "deployments apps/v1 in namespace shop", + "replicasets apps/v1 in namespace shop", + "events events.k8s.io/v1 in namespace shop", + "resourcequotas v1 in namespace shop", + "limitranges v1 in namespace shop", + ] + no_children = _plain(evidence_lines(_report(children_observation=None), NOW)) + assert len(no_children) == 4 + + +def test_sections_come_in_order_under_one_heading_each() -> None: + headings = [row.text for row in table_rows(_report(), NOW) if isinstance(row, Heading)] + + assert headings[0].startswith("Recorded creation failures — what the controller recorded") + assert headings[1].startswith("ResourceQuotas — current values; not a reservation") + assert headings[2].startswith("LimitRanges — current constraints; other admission plugins") + + +def test_a_failure_row_says_what_recorded_it_how_often_when_and_why() -> None: + (row, *_) = _rows(_report()) + + assert row.cells == ( + "ReplicaSet api-7d9f", + "12 times · last 2m ago", + "quota compute — quota exceeded", + EXCEEDED, + ) + assert row.destination == Destination("apps", "replicasets", "shop", "api-7d9f", "uid-rs") + + +def test_a_condition_failure_and_an_undated_one_are_labelled() -> None: + condition = replace( + FAILURE, subject=DEPLOYMENT, source=FailureSource.CONDITION, count=1, at=None + ) + + (row, *_) = _rows(_report(failures=(condition,))) + + assert row.cells[:2] == ("Deployment api (condition)", "once · undated") + assert row.destination == Destination("apps", "deployments", "shop", "api", "uid-api") + + +@pytest.mark.parametrize( + ("quotas", "observation", "expected"), + [ + ((), Basis.OBSERVED, "quota compute (no longer present) — quota exceeded"), + ((), Basis.FORBIDDEN, "quota compute (not in the quotas read) — quota exceeded"), + ], +) +def test_a_quota_the_message_names_but_the_list_lacks_is_said( + quotas: tuple[QuotaProjection, ...], observation: Basis, expected: str +) -> None: + report = _report( + quotas=quotas, quotas_observation=_seen(_source("", "resourcequotas"), observation) + ) + + (row, *_) = _rows(report) + + assert row.cells[2] == expected + + +def test_limit_range_and_other_rejections_are_attributed_only_as_far_as_the_message_goes() -> None: + limit = replace( + FAILURE, + rejection=classify( + 'pods "x" is forbidden: maximum cpu usage per Container is 1, but limit is 2' + ), + ) + webhook = replace(FAILURE, rejection=classify('admission webhook "p" denied the request: no')) + + cells = [ + row.cells[2] + for row in _rows(_report(failures=(limit, webhook), quotas=(), limit_ranges=())) + ] + + assert cells == [ + "a LimitRange in this namespace — above a LimitRange maximum", + "other rejection", + ] + + +def test_quota_rows_show_values_verbatim_with_markers_and_scopes() -> None: + rows = [ + row + for row in _rows(_report(failures=())) + if row.destination and row.destination.resource == "resourcequotas" + ] + + assert [row.cells for row in rows] == [ + ("compute", "compute: requests.cpu", "used 2 / hard 2", "at its limit"), + ("", "object counts: pods", "used 20 / hard 20", "at its limit"), + ("", "object counts: services", "used 1 / hard 5", ""), + ("", "scopes", "NotTerminating; PriorityClass In gold, silver", ""), + ] + assert {row.destination for row in rows} == { + Destination("", "resourcequotas", "shop", "compute", "uid-q") + } + + +def test_a_stale_quota_says_its_status_is_not_yet_updated() -> None: + stale = replace( + QUOTA, + entries=(QuotaEntry("pods", QuotaGroup.COUNT, "20", None, Usage.NOT_ESTABLISHED),), + scopes=(), + scope_selector=(), + stale=True, + ) + + rows = [row.cells for row in _rows(_report(failures=(), quotas=(stale,), limit_ranges=()))] + + assert rows == [ + ("compute", "object counts: pods", "used ? / hard 20", "status not yet updated") + ] + + +def test_limit_range_rows_list_each_constraint() -> None: + rows = [row.cells for row in _rows(_report(failures=(), quotas=()))] + + assert rows == [("defaults", "Container cpu", "default 500m · defaultRequest 100m · max 1", "")] + + +def test_empty_policies_get_a_row_saying_so() -> None: + empty_quota = replace(QUOTA, entries=(), scopes=(), scope_selector=()) + empty_range = replace(LIMITS, values=()) + + rows = [ + row.cells + for row in _rows(_report(failures=(), quotas=(empty_quota,), limit_ranges=(empty_range,))) + ] + + assert rows == [("compute", "no hard limits", "", ""), ("defaults", "no constraints", "", "")] + + +def test_a_section_without_rows_is_left_out() -> None: + headings = [ + r + for r in table_rows(_report(failures=(), quotas=(), limit_ranges=()), NOW) + if isinstance(r, Heading) + ] + + assert headings == [] + + +def test_notes_place_the_rejections_in_time_and_counters_in_context() -> None: + notes = _plain(report_notes(_report(), NOW)) + + assert ( + "The rejections were recorded at the times shown; the quotas and LimitRanges are as read now." + in notes + ) + assert "The counters are what the controller last reported, not a count of Pods." in notes + assert not any(n.startswith(("No ResourceQuota", "No LimitRange")) for n in notes) + + +def test_no_recorded_failure_never_rules_one_out() -> None: + available = _plain(report_notes(_report(failures=()), NOW)) + denied = _plain( + report_notes( + _report( + failures=(), + events_observation=_seen(_source("events.k8s.io", "events"), Basis.FORBIDDEN), + ), + NOW, + ) + ) + + assert ( + "No creation failure is recorded in the Events and conditions read. Events expire, so " + "this does not show that none happened." + ) in available + assert any( + n.startswith("Not established: whether a creation failure is recorded: your permissions") + for n in denied + ) + + +def test_absent_policies_are_said_only_from_a_conclusive_list() -> None: + empty = _plain(report_notes(_report(quotas=(), limit_ranges=()), NOW)) + truncated = _plain( + report_notes( + _report( + quotas=(), + quotas_observation=_seen(_source("", "resourcequotas"), Basis.TRUNCATED), + limit_ranges=(), + limit_ranges_observation=_seen(_source("", "limitranges"), Basis.FORBIDDEN), + ), + NOW, + ) + ) + + assert "No ResourceQuota in namespace shop." in empty + assert "No LimitRange in namespace shop." in empty + assert not any( + n.startswith("No ResourceQuota") or n.startswith("No LimitRange") for n in truncated + ) + assert ( + "Not established: whether every ResourceQuota is listed: stopped at the page limit" + in truncated + ) + assert any( + n.startswith("Not established: whether every LimitRange is listed: your permissions") + for n in truncated + ) + + +def test_a_stale_policy_list_no_longer_rules_anything_out() -> None: + later = NOW + timedelta(minutes=5) + + notes = _plain(report_notes(_report(quotas=(), limit_ranges=()), later)) + + assert "Not established: whether every ResourceQuota is listed: the list is stale" in notes + assert "Not established: whether every LimitRange is listed: the list is stale" in notes + + +def test_an_inconclusive_child_list_is_said() -> None: + report = _report(children_observation=_seen(_source("apps", "replicasets"), Basis.FORBIDDEN)) + + notes = _plain(report_notes(report, NOW)) + + assert any( + n.startswith("Not established: whether every ReplicaSet is listed: your permissions") + for n in notes + ) + + +def test_earlier_incarnations_are_counted() -> None: + notes = _plain(report_notes(_report(earlier_failures=2), NOW)) + + assert ( + "2 creation failures recorded about earlier incarnations of this Deployment or its " + "ReplicaSets are not shown." + ) in notes + job = _report( + target=CreationTarget("Job", "shop", "j"), earlier_failures=1, children_observation=None + ) + assert ( + "1 creation failures recorded about earlier incarnations of this Job are not shown." + ) in _plain(report_notes(job, NOW)) + + +def test_a_gone_or_recreated_workload_is_said() -> None: + gone = _report( + workload=None, + failures=(), + workload_observation=_seen(WORKLOAD_SOURCE, Basis.OBJECT_ABSENT), + ) + recreated = _report(workload_observation=_seen(WORKLOAD_SOURCE, uid="uid-new")) + + assert "The Deployment no longer exists." in _plain(report_notes(gone, NOW)) + assert not any("counters" in n for n in _plain(report_notes(gone, NOW))) + assert ( + "The Deployment did not exist when it was read; it may have been created since." + in _plain(report_notes(gone, NOW + timedelta(minutes=5))) + ) + assert ( + "The Deployment was recreated since it was selected: only the new one's evidence is shown." + ) in _plain(report_notes(recreated, NOW)) + + +class _Host(App[None]): + def __init__(self, screen: CreationConstraintsScreen) -> None: + super().__init__() + self.prompt = screen + self.result: list[Destination | None] = [] + + def on_mount(self) -> None: + self.push_screen(self.prompt, self.result.append) + + +async def test_enter_on_a_row_returns_where_it_leads() -> None: + app = _Host(CreationConstraintsScreen(_report(), lambda: NOW)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + table = app.screen.query_one(DataTable) + assert table.cursor_row == 1 + await pilot.press("down", "down", "enter") + await until(pilot, lambda: bool(app.result), label="view closed") + + assert app.result == [Destination("", "resourcequotas", "shop", "compute", "uid-q")] + + +async def test_enter_on_a_heading_does_nothing_and_escape_closes() -> None: + app = _Host(CreationConstraintsScreen(_report(), lambda: NOW)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + await pilot.press("up", "enter") + await pilot.pause() + assert app.result == [] + await pilot.press("escape") + await until(pilot, lambda: bool(app.result), label="view closed") + + assert app.result == [None] + + +async def test_cluster_text_is_shown_literally() -> None: + hostile = replace(FAILURE, rejection=classify("[bold]boom[/bold]")) + target = CreationTarget("Deployment", "shop", "[red]api[/red]") + app = _Host( + CreationConstraintsScreen( + _report(target=target, failures=(hostile,), workload=None), lambda: NOW + ) + ) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + title = app.screen.query_one("#creation-title", Static) + assert "[red]api[/red]" in str(title.render()) + table = app.screen.query_one(DataTable) + assert str(table.get_row_at(1)[3]) == "[bold]boom[/bold]" + + +async def test_an_open_view_ages_its_evidence_without_reading_again() -> None: + now = [NOW] + app = _Host(CreationConstraintsScreen(_report(quotas=()), lambda: now[0], refresh_seconds=0.05)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + evidence = app.screen.query_one("#creation-evidence", Static) + notes = app.screen.query_one("#creation-notes", Static) + assert "No ResourceQuota in namespace shop." in str(notes.render()) + + now[0] = NOW + timedelta(minutes=5) + await until(pilot, lambda: "· stale" in str(evidence.render()), label="evidence aged") + + assert "No ResourceQuota in namespace shop." not in str(notes.render()) + + +def test_without_the_workload_there_are_no_counters_to_explain() -> None: + denied = _report(workload=None, workload_observation=_seen(WORKLOAD_SOURCE, Basis.FORBIDDEN)) + + assert not any("counters" in n for n in _plain(report_notes(denied, NOW))) + + +def test_a_policy_that_could_not_be_read_in_full_never_claims_to_be_empty() -> None: + quota = replace(QUOTA, entries=(), scopes=(), scope_selector=(), complete=False) + partial = replace(QUOTA, name="partial", scopes=(), scope_selector=(), complete=False) + limit_range = replace(LIMITS, values=(), complete=False) + + rows = [ + row.cells + for row in _rows(_report(failures=(), quotas=(quota, partial), limit_ranges=(limit_range,))) + ] + + assert ("compute", "constraints could not be read", "", "") in rows + assert ("", "constraints could not be read", "", "") in rows + assert ("defaults", "constraints could not be read", "", "") in rows + assert not any(cells[1] in ("no hard limits", "no constraints") for cells in rows) + + +async def test_an_open_view_ages_its_table_cells_too() -> None: + now = [NOW] + report = _report(quotas=(), limit_ranges=()) + app = _Host(CreationConstraintsScreen(report, lambda: now[0], refresh_seconds=0.05)) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + table = app.screen.query_one(DataTable) + assert str(table.get_row_at(1)[1]) == "12 times · last 2m ago" + assert str(table.get_row_at(1)[2]) == "quota compute (no longer present) — quota exceeded" + + now[0] = NOW + timedelta(minutes=5) + await until( + pilot, + lambda: str(table.get_row_at(1)[1]) == "12 times · last 7m ago", + label="age refreshed", + ) + + assert str(table.get_row_at(1)[2]) == ( + "quota compute (not in the quotas read) — quota exceeded" + ) + assert table.cursor_row == 1 + + +async def test_a_multiline_message_is_shown_in_full() -> None: + multiline = replace(FAILURE, rejection=classify("first line\nsecond line\nthird line")) + app = _Host( + CreationConstraintsScreen( + _report(failures=(multiline,), quotas=(), limit_ranges=()), lambda: NOW + ) + ) + async with app.run_test() as pilot: + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + table = app.screen.query_one(DataTable) + row_key = table.coordinate_to_cell_key(Coordinate(1, 0)).row_key + assert table.rows[row_key].height == 3 + assert str(table.get_row_at(1)[3]) == "first line\nsecond line\nthird line" + + +def test_an_unestablished_comparison_is_marked_not_left_blank() -> None: + unknown = replace( + QUOTA, + entries=(QuotaEntry("pods", QuotaGroup.COUNT, "20", "lots", Usage.NOT_ESTABLISHED),), + scopes=(), + scope_selector=(), + ) + + rows = [row.cells for row in _rows(_report(failures=(), quotas=(unknown,), limit_ranges=()))] + + assert rows == [("compute", "object counts: pods", "used lots / hard 20", "not established")] + + +def test_no_recorded_failure_names_only_the_sources_actually_read() -> None: + unread_children = _report( + failures=(), children_observation=_seen(_source("apps", "replicasets"), Basis.FORBIDDEN) + ) + unread_workload = _report( + failures=(), + workload=None, + workload_observation=_seen(WORKLOAD_SOURCE, Basis.FORBIDDEN), + ) + + for report in (unread_children, unread_workload): + notes = _plain(report_notes(report, NOW)) + assert ( + "No creation failure is recorded in the Events read. Events expire, so this does " + "not show that none happened." + ) in notes + assert ( + "Not established: whether a failure is recorded about the Deployment or its " + "ReplicaSets: they could not all be read" + ) in notes + assert not any("Events and conditions read" in n for n in notes) + + +def test_a_kind_without_replica_failure_conditions_names_only_events() -> None: + job = _report(target=CreationTarget("Job", "shop", "j"), failures=(), children_observation=None) + + assert ( + "No creation failure is recorded in the Events read. Events expire, so this does not " + "show that none happened." + ) in _plain(report_notes(job, NOW)) diff --git a/tests/ui/test_creation_constraints_workflow.py b/tests/ui/test_creation_constraints_workflow.py new file mode 100644 index 000000000..faaa11c0a --- /dev/null +++ b/tests/ui/test_creation_constraints_workflow.py @@ -0,0 +1,204 @@ +"""`:blocked` end to end: a Deployment row, its recorded failures, and their ReplicaSet (#411).""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from datetime import UTC, datetime +from typing import Any + +from korvid.core.config import KorvidConfig +from korvid.core.store import ResourceStore, Summary +from korvid.core.watch import WatchManager +from korvid.k8s.discovery import PODS_META, ResourceMeta, build_alias_map +from korvid.k8s.discovery_snapshot import ( + GroupVersionOutcome, + RawDiscovery, + RootOutcome, + ServedResource, +) +from korvid.k8s.evidence_requests import EvidenceClient, EvidencePage, EvidenceTarget +from korvid.k8s.models import GenericSummary +from korvid.ui.widgets.creation_constraints_screen import CreationConstraintsScreen +from tests.app_factory import build_test_app +from tests.ui.waits import screen_ready, until + +DEPLOYMENTS = ResourceMeta("Deployment", "deployments", "apps", "v1", True) +REPLICA_SETS = ResourceMeta("ReplicaSet", "replicasets", "apps", "v1", True) +EXCEEDED = 'pods "api-7d9f-x" is forbidden: exceeded quota: compute, requested: pods=1' +_SERVED = { + ("", "v1"): ("resourcequotas", "limitranges", "events"), + ("apps", "v1"): ("deployments", "replicasets"), +} + + +class Client(EvidenceClient): + """Serves one Deployment whose ReplicaSet records a quota rejection.""" + + def __init__(self) -> None: + self.reads: list[tuple[str, str, str]] = [] + + @property + def session_generation(self) -> int: + return 1 + + async def discovery_snapshot(self) -> RawDiscovery: + outcomes = tuple( + GroupVersionOutcome( + group, version, tuple(ServedResource(r, frozenset({"get", "list"})) for r in names) + ) + for (group, version), names in _SERVED.items() + ) + return RawDiscovery( + 1, + datetime.now(UTC), + 0.0, + RootOutcome("/api", (("", "v1"),)), + RootOutcome("/apis", (("apps", "v1"),)), + outcomes, + ) + + async def list_evidence_page( + self, target: EvidenceTarget, *, limit: int, continuation: str, max_bytes: int + ) -> EvidencePage: + self.reads.append(("list", target.resource, target.namespace or "")) + if target.resource == "replicasets": + owner = [{"uid": "uid-api", "controller": True}] + rs = { + "metadata": { + "name": "api-7d9f", + "namespace": "default", + "uid": "uid-rs", + "ownerReferences": owner, + } + } + return EvidencePage((rs,)) + if target.resource == "events": + event = { + "metadata": {"name": "e", "namespace": "default", "uid": "uid-e"}, + "reason": "FailedCreate", + "involvedObject": {"kind": "ReplicaSet", "name": "api-7d9f", "uid": "uid-rs"}, + "message": EXCEEDED, + } + return EvidencePage((event,)) + if target.resource == "resourcequotas": + quota = { + "metadata": {"name": "compute", "namespace": "default", "uid": "uid-q"}, + "spec": {"hard": {"pods": "0"}}, + "status": {"hard": {"pods": "0"}, "used": {"pods": "0"}}, + } + return EvidencePage((quota,)) + return EvidencePage(()) + + async def get_evidence_object( + self, target: EvidenceTarget, name: str, *, max_bytes: int + ) -> dict[str, Any]: + self.reads.append(("get", target.resource, name)) + template = { + "spec": { + "containers": [{"name": "app", "env": [{"name": "T", "value": "s3cr3t-literal"}]}] + } + } + return { + "metadata": {"name": name, "namespace": "default", "uid": "uid-api"}, + "spec": {"replicas": 1, "template": template}, + "status": {}, + } + + +async def _source(kind: str, scope: str) -> AsyncIterator[tuple[str, Summary]]: + if kind == "deployments": + yield ("ADDED", GenericSummary("api", "default", "Deployment", "", uid="uid-api")) + if kind == "replicasets": + yield ("ADDED", GenericSummary("api-7d9f", "default", "ReplicaSet", "", uid="uid-rs")) + await asyncio.Event().wait() + + +async def _command(pilot: Any, text: str) -> None: + await pilot.press("colon", *text, "enter") + + +def _app(client: EvidenceClient) -> Any: + store = ResourceStore() + return build_test_app( + config=KorvidConfig(namespace="default"), + store=store, + watch_manager=WatchManager(store, _source), + aliases=build_alias_map([PODS_META, DEPLOYMENTS, REPLICA_SETS]), + agent_available=False, + evidence_client=client, + ) + + +async def test_a_deployments_failures_lead_to_their_replica_set() -> None: + client = Client() + app = _app(client) + async with app.run_test() as pilot: + await _command(pilot, "deployments") + await until( + pilot, + lambda: app._view.selected_ns_name(notify=False) == ("default", "api"), + label="Deployment selected", + ) + + await _command(pilot, "blocked") + await until(pilot, lambda: screen_ready(app, CreationConstraintsScreen), label="view open") + screen = app.screen + assert isinstance(screen, CreationConstraintsScreen) + (failure,) = screen.report.failures + assert failure.rejection.quotas == ("compute",) + assert "s3cr3t-literal" not in repr(screen.report) + assert sorted(client.reads) == [ + ("get", "deployments", "api"), + ("list", "events", "default"), + ("list", "limitranges", "default"), + ("list", "replicasets", "default"), + ("list", "resourcequotas", "default"), + ] + + await pilot.press("enter") + await until(pilot, lambda: app.current_kind == "replicasets", label="ReplicaSet view") + await until( + pilot, + lambda: app._view.selected_ns_name(notify=False) == ("default", "api-7d9f"), + label="ReplicaSet selected", + ) + + +async def test_blocked_on_another_view_says_what_to_open() -> None: + app = _app(Client()) + async with app.run_test() as pilot: + await _command(pilot, "blocked") + await until( + pilot, + lambda: any("Open Deployments, ReplicaSets" in n.message for n in app._notifications), + label="reason shown", + ) + assert not isinstance(app.screen, CreationConstraintsScreen) + + +class BrokenClient(Client): + async def discovery_snapshot(self) -> RawDiscovery: + raise RuntimeError("discovery exploded") + + +async def test_a_failed_run_is_reported_instead_of_vanishing() -> None: + app = _app(BrokenClient()) + async with app.run_test() as pilot: + await _command(pilot, "deployments") + await until( + pilot, + lambda: app._view.selected_ns_name(notify=False) == ("default", "api"), + label="Deployment selected", + ) + await _command(pilot, "blocked") + await until( + pilot, + lambda: any( + "Workload creation failure explanation failed" in n.message + and "discovery exploded" in n.message + for n in app._notifications + ), + label="failure reported", + ) + assert app.is_running diff --git a/tests/ui/test_evidence_text.py b/tests/ui/test_evidence_text.py index e5ebc7911..48333bbad 100644 --- a/tests/ui/test_evidence_text.py +++ b/tests/ui/test_evidence_text.py @@ -22,6 +22,7 @@ from korvid.ui.evidence_text import ( age_text, basis_sentence, + elapsed_text, not_established, observation_line, ) @@ -161,3 +162,11 @@ def test_not_established_says_why() -> None: assert line.plain == "Not established: whether every consumer is listed" assert not_established("[red]x").plain == "Not established: [red]x" + + +@pytest.mark.parametrize( + ("seconds", "text"), + [(0.5, "just now"), (120, "2m ago"), (-5, "in the future")], +) +def test_an_elapsed_time_reads_without_a_verb(seconds: float, text: str) -> None: + assert elapsed_text(NOW - timedelta(seconds=seconds), NOW) == text diff --git a/tests/ui/test_session_timeline_controller.py b/tests/ui/test_session_timeline_controller.py index 82abac311..b538c7811 100644 --- a/tests/ui/test_session_timeline_controller.py +++ b/tests/ui/test_session_timeline_controller.py @@ -54,6 +54,7 @@ def __init__(self, *, close_worker_coroutines: bool = True) -> None: self.notifications: list[_Notification] = [] self.workers: list[Any] = [] # stores the coroutine/callable passed to run_worker self.worker_groups: list[tuple[str, bool]] = [] # (group, exit_on_error) + self.exclusive_workers: list[bool] = [] self.cancelled_groups: list[str] = [] self.screens: list[Any] = [] self.screen_callback: Callable[[Any], None] | None = None @@ -103,6 +104,7 @@ def run_worker( else: self.workers.append(work) self.worker_groups.append((group, exit_on_error)) + self.exclusive_workers.append(exclusive) return None async def cancel_workers(self, group: str) -> None: diff --git a/tests/ui/test_worker_failures.py b/tests/ui/test_worker_failures.py new file mode 100644 index 000000000..ee04e0cd2 --- /dev/null +++ b/tests/ui/test_worker_failures.py @@ -0,0 +1,12 @@ +"""Which worker failures the app reports instead of crashing.""" + +from __future__ import annotations + +from korvid.ui.creation_constraints_controller import CREATION_CONSTRAINTS_GROUP +from korvid.ui.worker_failures import REPORTED_WORKER_GROUPS + + +def test_a_failed_creation_constraints_run_is_reported() -> None: + assert REPORTED_WORKER_GROUPS[CREATION_CONSTRAINTS_GROUP] == ( + "Workload creation failure explanation failed" + ) diff --git a/tests/version_matrix/fixtures.yaml b/tests/version_matrix/fixtures.yaml index 2747bf2f9..c6d7a3e1d 100644 --- a/tests/version_matrix/fixtures.yaml +++ b/tests/version_matrix/fixtures.yaml @@ -187,3 +187,57 @@ subjects: - kind: ServiceAccount name: events-group namespace: korvid-matrix +--- +# Workload creation constraints (#411). A namespace of its own, because its +# quota admits no Pod at all: the Deployment's ReplicaSet records the +# rejection as FailedCreate. The pause image is already on every kind node. +apiVersion: v1 +kind: Namespace +metadata: + name: korvid-matrix-quota +--- +apiVersion: v1 +kind: ResourceQuota +metadata: + name: matrix-pods + namespace: korvid-matrix-quota +spec: + hard: + pods: "0" +--- +apiVersion: v1 +kind: LimitRange +metadata: + name: matrix-limits + namespace: korvid-matrix-quota +spec: + limits: + - type: Container + default: + cpu: 200m + defaultRequest: + cpu: 100m + max: + cpu: "1" +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: blocked + namespace: korvid-matrix-quota +spec: + replicas: 1 + selector: + matchLabels: + app: blocked + template: + metadata: + labels: + app: blocked + spec: + containers: + - name: app + image: registry.k8s.io/pause:3.10 + env: + - name: LITERAL + value: matrix-template-literal diff --git a/tests/version_matrix/test_scenarios.py b/tests/version_matrix/test_scenarios.py index 123daca54..5eb2c7d13 100644 --- a/tests/version_matrix/test_scenarios.py +++ b/tests/version_matrix/test_scenarios.py @@ -8,6 +8,7 @@ from __future__ import annotations +import asyncio import json from datetime import UTC, datetime @@ -25,6 +26,8 @@ ) from korvid.core.config_consumers import ConsumptionMode from korvid.core.config_propagation import ConfigTarget, explain_propagation +from korvid.core.creation_constraints import CreationReport, CreationTarget, explain_creation +from korvid.core.creation_failures import RejectionKind from korvid.core.evidence_reads import ( LAST_APPLIED_ANNOTATION, read_list, @@ -267,3 +270,42 @@ async def test_a_config_maps_consumers_are_read_by_mode(connect: Connect) -> Non assert report.consumers_conclusive(datetime.now(UTC)) assert "matrix-configmap-payload" not in repr(report) assert "matrix-pod-literal" not in repr(report) + + +QUOTA_NAMESPACE = "korvid-matrix-quota" +#: How long the ReplicaSet controller gets to record its first rejection. +_REJECTION_WAIT = 90.0 + + +async def _recorded(connect: Connect) -> CreationReport: + """The blocked Deployment's report, once a rejection and current quota status are in.""" + kube = await connect(ADMIN_CONTEXT) + target = CreationTarget("Deployment", QUOTA_NAMESPACE, "blocked") + deadline = asyncio.get_running_loop().time() + _REJECTION_WAIT + while True: + report = await explain_creation(kube, await snapshot_of(kube), target) + settled = report.failures and report.quotas and not report.quotas[0].stale + if settled or asyncio.get_running_loop().time() > deadline: + return report + await asyncio.sleep(2) + + +async def test_a_quota_rejection_is_read_next_to_the_namespace_policies(connect: Connect) -> None: + report = await _recorded(connect) + + assert _state(report.workload_observation) == (EvidenceState.AVAILABLE, Basis.OBSERVED) + assert [child.subject.kind for child in report.children] == ["ReplicaSet"] + assert report.failures, "the ReplicaSet recorded no FailedCreate in time" + quota_kinds = {RejectionKind.QUOTA_EXCEEDED, RejectionKind.QUOTA_STATUS_UNKNOWN} + assert any( + quota_kinds & set(f.rejection.kinds) and f.rejection.quotas == ("matrix-pods",) + for f in report.failures + ) + (quota,) = report.quotas + assert (quota.name, [(e.resource, e.hard) for e in quota.entries]) == ( + "matrix-pods", + [("pods", "0")], + ) + assert [r.name for r in report.limit_ranges] == ["matrix-limits"] + assert report.quotas_conclusive(datetime.now(UTC)) + assert "matrix-template-literal" not in repr(report)