discovery block for expired licenses

Author: Redbot

src/redfetch/sync_discovery.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import os
+import time
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -91,6 +92,7 @@ class _RootSpec:
     """Tracks why a resource was selected for sync and its raw API payload."""
     sources: set[str] = field(default_factory=set)
     payload: dict | None = None
+    discovery_block: str | None = None
 
 
 def payload_title(payload: dict | None) -> str | None:
@@ -149,13 +151,17 @@ def _root_sources_for_full_sync(
     for license_info in licenses:
         if not license_info.get("active", False):
             continue
+        end_date = license_info.get("end_date", 0)
+        is_expired = end_date != 0 and end_date < time.time()
         payload = license_info.get("resource") or {}
         category_id = payload_category_id(payload)
         if not _category_allowed_in_env(category_id, settings_env):
             continue
         resource_id = str(payload["resource_id"])
         spec = specs.setdefault(resource_id, _RootSpec())
         spec.sources.add("licensed")
+        if is_expired:
+            spec.discovery_block = "license_expired"
         spec.payload = payload
 
     settings_for_env = config.settings.from_env(settings_env)
@@ -313,6 +319,8 @@ async def discover_desired_set(
             payload=spec.payload,
             settings_env=settings_env,
         )
+        if spec.discovery_block:
+            root_target.discovery_block = spec.discovery_block
         _expand_dependencies(
             desired_set,
             parent_target=root_target,

src/redfetch/sync_planner.py

@@ -116,6 +116,8 @@ def _decide_action(
         parent_action is None or parent_action.action in {"block", "untrack"}
     ):
         return "block", "parent_blocked"
+    if target.discovery_block:
+        return "block", target.discovery_block
     if remote_state is None:
         return "block", "fetch_error"
     if remote_state.status in BLOCKING_STATUSES:

src/redfetch/sync_types.py

@@ -35,6 +35,7 @@
     "parent_failed",
     "dependency_cycle",
     "unknown_category",
+    "license_expired",
 ]
 ResultOutcome = Literal["downloaded", "skipped", "blocked", "untracked", "error"]
 
@@ -62,6 +63,7 @@ class ReasonInfo:
     "not_installed":           ReasonInfo("Not yet installed locally."),
     "already_current":         ReasonInfo("Already up to date."),
     "install_context_changed": ReasonInfo("Install location or settings changed; re-downloading."),
+    "license_expired":         ReasonInfo("Your license for this resource has expired.", quiet=True, summary_label="Licenses expired"),
 }
 
 
@@ -128,6 +130,7 @@ class DesiredInstallTarget(TargetIdentity):
     flatten: bool = False
     protected_files: list[str] = Field(default_factory=list)
     explicit_root: bool = False
+    discovery_block: PlanReason | None = None
 
 
 class DesiredSet(SyncModel):
@@ -154,6 +157,7 @@ def add_target(self, target: DesiredInstallTarget) -> DesiredInstallTarget:
         existing.flatten = existing.flatten or target.flatten
         existing.protected_files = existing.protected_files or target.protected_files
         existing.explicit_root = existing.explicit_root or target.explicit_root
+        existing.discovery_block = existing.discovery_block or target.discovery_block
         return existing
 
     def resource_targets(self, resource_id: str) -> list[DesiredInstallTarget]:

tests/test_licensed_resources_filtering.py

@@ -1,13 +1,26 @@
 """Discovery-stage tests for licensed resource filtering."""
 
 import asyncio
+import time
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
 import pytest
 
 from redfetch import sync_discovery as discovery
+from redfetch import sync_planner as planner
+from redfetch.sync_types import (
+    DesiredInstallTarget,
+    DesiredSet,
+    LocalSnapshot,
+    RemoteArtifact,
+    RemoteResourceState,
+    RemoteSnapshot,
+)
+
+FAR_FUTURE = int(time.time()) + 86400 * 365 * 10
+PAST = int(time.time()) - 86400 * 30
 
 
 def make_license(
@@ -17,11 +30,12 @@ def make_license(
     *,
     version_id: int = 101,
     file_id: int = 1001,
+    end_date: int = FAR_FUTURE,
 ) -> dict:
     return {
         "active": True,
-        "start_date": "2024-01-01",
-        "end_date": "2025-01-01",
+        "start_date": 1711170000,
+        "end_date": end_date,
         "license_id": 12345,
         "resource": {
             "resource_id": resource_id,
@@ -109,3 +123,116 @@ def test_cross_compatible_licensed_resources_remain_in_scope(env, category_id, r
     target_key = f"/{resource_id}/"
     assert target_key in desired_set.install_targets
     assert desired_set.install_targets[target_key].sources == {"licensed"}
+
+
+# --- expired license discovery tests ---
+
+
+def test_expired_license_gets_discovery_block():
+    desired_set = asyncio.run(
+        _discover_from_licenses([make_license(9998, 8, end_date=PAST)], "LIVE")
+    )
+    target = desired_set.install_targets["/9998/"]
+    assert target.sources == {"licensed"}
+    assert target.discovery_block == "license_expired"
+
+
+def test_valid_license_has_no_discovery_block():
+    desired_set = asyncio.run(
+        _discover_from_licenses([make_license(9998, 8, end_date=FAR_FUTURE)], "LIVE")
+    )
+    target = desired_set.install_targets["/9998/"]
+    assert target.sources == {"licensed"}
+    assert target.discovery_block is None
+
+
+def test_unlimited_license_has_no_discovery_block():
+    desired_set = asyncio.run(
+        _discover_from_licenses([make_license(9998, 8, end_date=0)], "LIVE")
+    )
+    target = desired_set.install_targets["/9998/"]
+    assert target.sources == {"licensed"}
+    assert target.discovery_block is None
+
+
+# --- expired license planner tests ---
+
+
+def _downloadable_state(resource_id: str, *, category_id: int = 8) -> RemoteResourceState:
+    return RemoteResourceState(
+        resource_id=resource_id,
+        title=f"Resource {resource_id}",
+        category_id=category_id,
+        version_id=1234,
+        status="downloadable",
+        artifact=RemoteArtifact(
+            file_id=9876,
+            filename=f"{resource_id}.zip",
+            download_url=f"https://example.com/{resource_id}.zip",
+            file_hash="d41d8cd98f00b204e9800998ecf8427e",
+        ),
+        source_note="manifest_plus_access_check",
+    )
+
+
+def _build_plan_for_target(target, remote_state):
+    mock_settings = MagicMock()
+    mock_settings.ENV = "LIVE"
+    mock_settings.from_env.return_value = SimpleNamespace(
+        DOWNLOAD_FOLDER="C:/downloads",
+        SPECIAL_RESOURCES={},
+        PROTECTED_FILES_BY_RESOURCE={},
+    )
+    desired_set = DesiredSet(
+        mode="full",
+        resource_ids={target.resource_id},
+        install_targets={target.target_key: target},
+    )
+    with patch("redfetch.sync_planner.config.settings", mock_settings), \
+         patch("redfetch.sync_planner.config.CATEGORY_MAP", {8: "macros", 11: "plugins", 25: "lua"}):
+        return planner.build_execution_plan(
+            desired_set=desired_set,
+            remote_snapshot=RemoteSnapshot(resources={target.resource_id: remote_state}),
+            local_snapshot=LocalSnapshot(),
+            settings_env="LIVE",
+        )
+
+
+def test_planner_blocks_on_discovery_block():
+    target = DesiredInstallTarget(
+        target_key="/9998/",
+        resource_id="9998",
+        parent_id=None,
+        parent_target_key=None,
+        root_resource_id="9998",
+        target_kind="root",
+        sources={"licensed"},
+        title="Expired Resource",
+        category_id=8,
+        resolved_path="C:/downloads/9998",
+        discovery_block="license_expired",
+    )
+    plan = _build_plan_for_target(target, _downloadable_state("9998"))
+    action = plan.actions["/9998/"]
+    assert action.action == "block"
+    assert action.reason == "license_expired"
+
+
+def test_planner_blocks_discovery_block_even_when_watching():
+    target = DesiredInstallTarget(
+        target_key="/9998/",
+        resource_id="9998",
+        parent_id=None,
+        parent_target_key=None,
+        root_resource_id="9998",
+        target_kind="root",
+        sources={"watching", "licensed"},
+        title="Watched + Expired Resource",
+        category_id=8,
+        resolved_path="C:/downloads/9998",
+        discovery_block="license_expired",
+    )
+    plan = _build_plan_for_target(target, _downloadable_state("9998"))
+    action = plan.actions["/9998/"]
+    assert action.action == "block"
+    assert action.reason == "license_expired"