Archived
forge-recoveryandClaude Sonnet 5.5
530c3639d1
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
504 lines
31 KiB
Python
504 lines
31 KiB
Python
"""Source-only tests of the effective version strategy. UNTESTED: written while test execution was suspended; never executed."""
|
|
|
|
import base64
|
|
import io
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.parse
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from datetime import UTC, datetime
|
|
from email.message import Message
|
|
from pathlib import Path
|
|
from typing import cast, override
|
|
|
|
from platform_ci.build_inputs import compute_build_inputs
|
|
from platform_ci.configuration import ConfigurationError, ProjectConfiguration, parse_project_configuration
|
|
from platform_ci.delivery.errors import DeliveryError
|
|
from platform_ci.delivery.forge import ForgeGitHistory, RepositoryScope, prepare_job_started
|
|
from platform_ci.delivery.git_history import GitHistoryError
|
|
from platform_ci.delivery.publisher import _plan_for_run # pyright: ignore[reportPrivateUsage]
|
|
from platform_ci.delivery.selection import release_activity_for_run
|
|
from platform_ci.effective_version import (
|
|
DEVELOP_NEXT_MINOR,
|
|
FEATURE_SNAPSHOT,
|
|
PATCH_BUMP,
|
|
RC_BASELINE_NONE,
|
|
RC_BASELINE_RECONCILED,
|
|
RC_FROM_BRANCH,
|
|
RELEASE_ACTIVITY_WINDOW_SECONDS,
|
|
STABLE_FROM_TAG,
|
|
ReleaseActivityError,
|
|
effective_version,
|
|
has_release_activity,
|
|
message_matches_release_activity,
|
|
release_activity_since,
|
|
version_parts,
|
|
with_version_property,
|
|
)
|
|
from platform_ci.event_selection import EventFacts, push_facts, tag_facts
|
|
from platform_ci.plan import ReleaseActivityReader, Target, plan_targets
|
|
from platform_ci.versions import parse_declared_version
|
|
from platform_ci.workspace import PipelineError
|
|
from tests.support import SOURCE_REVISION, SOURCE_TREE, TEST_RUNTIME_CATALOG, component_mapping, example_mapping
|
|
|
|
POM = """<project xmlns="http://maven.apache.org/POM/4.0.0">
|
|
<modelVersion>4.0.0</modelVersion>
|
|
<groupId>example</groupId>
|
|
<artifactId>library</artifactId>
|
|
<version>${{revision}}</version>
|
|
<properties><revision>{revision}</revision></properties>
|
|
</project>
|
|
"""
|
|
LITERAL_POM = POM.replace("${{revision}}", "9.9.9")
|
|
|
|
|
|
def _configuration(rule: str = RC_FROM_BRANCH, trigger: Mapping[str, object] | None = None, extra: dict[str, object] | None = None, publish: bool = True) -> dict[str, object]:
|
|
"""The Maven publication example with one candidate channel carrying an effective version rule (and, unless `publish` is false, one Maven publication on it)."""
|
|
configuration = example_mapping("maven-publication.ci.yaml")
|
|
channel: dict[str, object] = {"trigger": trigger or {"push_branches": ["release/**"]}, "versioning": "maven-release", "effective_version": {"rule": rule}}
|
|
channel.update(extra or {})
|
|
publications = {"library-packages": {"component": "library", "artifact": "packages", "destination": "company-packages", "channels": ["candidate"]}} if publish else {}
|
|
configuration["release"] = {"channels": {"candidate": channel}, "publications": publications}
|
|
return configuration
|
|
|
|
|
|
def _reader(revision: str, pom: str = POM) -> Callable[[Target], str]:
|
|
"""A declared-version reader over a fixed POM, the same parser the checkout and forge readers use."""
|
|
return lambda target: parse_declared_version(target["build_tool"], pom.format(revision=revision).encode(), target["commands"], target.get("version_source"))
|
|
|
|
|
|
def _plan(project: ProjectConfiguration, facts: EventFacts, revision: str = "1.5.0-SNAPSHOT", pom: str = POM, release_activity: ReleaseActivityReader | None = None) -> list[Target]:
|
|
return plan_targets(project, TEST_RUNTIME_CATALOG, facts, _reader(revision, pom), release_activity)
|
|
|
|
|
|
class RuleTest(unittest.TestCase):
|
|
def test_release_branch_first_candidate(self) -> None:
|
|
self.assertEqual(effective_version(RC_FROM_BRANCH, "1.5.0-SNAPSHOT", "release/1.6", ""), "1.6.0-RC.1")
|
|
|
|
def test_release_branch_increments_a_declared_candidate(self) -> None:
|
|
self.assertEqual(effective_version(RC_FROM_BRANCH, "1.6.0-RC.3", "release/1.6", ""), "1.6.0-RC.4")
|
|
|
|
def test_release_branch_accepts_every_captured_form(self) -> None:
|
|
"""The captured search is unanchored: any `release/<digits>.<digits>` prefix works, as the legacy qbit and core RC branches used."""
|
|
for branch in ("release/1.6", "release/1.6.0", "release/1.6.x", "release/1.6-hotfix"):
|
|
self.assertEqual(effective_version(RC_FROM_BRANCH, "1.5.0-SNAPSHOT", branch, ""), "1.6.0-RC.1")
|
|
|
|
def test_release_branch_without_a_numeric_line_is_refused(self) -> None:
|
|
for branch in ("release/next", "develop", ""):
|
|
with self.assertRaises(PipelineError):
|
|
effective_version(RC_FROM_BRANCH, "1.5.0-SNAPSHOT", branch, "")
|
|
|
|
def test_branch_rules_refuse_a_tag(self) -> None:
|
|
for rule in (DEVELOP_NEXT_MINOR, RC_FROM_BRANCH, PATCH_BUMP):
|
|
with self.assertRaises(PipelineError):
|
|
effective_version(rule, "1.5.0", "", "v1.5.0")
|
|
|
|
def test_develop_next_minor_from_rc_or_stable(self) -> None:
|
|
for declared in ("1.5.0-RC.2", "1.5.0"):
|
|
self.assertEqual(effective_version(DEVELOP_NEXT_MINOR, declared, "develop", ""), "1.6.0-SNAPSHOT")
|
|
with self.assertRaises(PipelineError):
|
|
effective_version(DEVELOP_NEXT_MINOR, "1.5.0", "dev", "")
|
|
|
|
def test_develop_with_a_snapshot_and_no_release_activity_input_is_refused(self) -> None:
|
|
"""Without the history input the captured handler's outcome for a SNAPSHOT is unknown; it is refused, never assumed to be `keep`."""
|
|
for declared in ("1.5.0-SNAPSHOT", "1.5.0-new-abc1234-SNAPSHOT"):
|
|
with self.assertRaisesRegex(ReleaseActivityError, "merged into develop within 3 days"):
|
|
effective_version(DEVELOP_NEXT_MINOR, declared, "develop", "")
|
|
|
|
def test_develop_snapshot_is_unchanged_without_release_activity(self) -> None:
|
|
"""The captured `keep` outcome: the declared SNAPSHOT is the build version, plain or feature-format."""
|
|
for declared in ("0.5.0-SNAPSHOT", "1.5.0-new-abc1234-SNAPSHOT"):
|
|
self.assertEqual(effective_version(DEVELOP_NEXT_MINOR, declared, "develop", "", release_activity=lambda: False), declared)
|
|
|
|
def test_develop_snapshot_bumps_to_the_next_minor_after_recent_release_activity(self) -> None:
|
|
self.assertEqual(effective_version(DEVELOP_NEXT_MINOR, "0.5.0-SNAPSHOT", "develop", "", release_activity=lambda: True), "0.6.0-SNAPSHOT")
|
|
self.assertEqual(effective_version(DEVELOP_NEXT_MINOR, "1.5.0-new-abc1234-SNAPSHOT", "develop", "", release_activity=lambda: True), "1.6.0-SNAPSHOT")
|
|
|
|
def test_release_activity_is_asked_only_when_it_decides_the_version_and_only_once(self) -> None:
|
|
calls: list[str] = []
|
|
|
|
def activity() -> bool:
|
|
calls.append("asked")
|
|
return False
|
|
|
|
for declared in ("1.5.0-RC.2", "1.5.0"): # RC and stable always bump: the captured handler never looks at history
|
|
self.assertEqual(effective_version(DEVELOP_NEXT_MINOR, declared, "develop", "", release_activity=activity), "1.6.0-SNAPSHOT")
|
|
self.assertEqual(calls, [])
|
|
effective_version(DEVELOP_NEXT_MINOR, "1.5.0-SNAPSHOT", "develop", "", release_activity=activity)
|
|
self.assertEqual(calls, ["asked"])
|
|
|
|
def test_reconciled_rc_baseline_refuses_only_the_undecidable_first_counter(self) -> None:
|
|
with self.assertRaisesRegex(PipelineError, "tagged first-parent RC counter"):
|
|
effective_version(RC_FROM_BRANCH, "1.6.0-SNAPSHOT", "release/1.6", "", RC_BASELINE_RECONCILED)
|
|
self.assertEqual(effective_version(RC_FROM_BRANCH, "1.6.0-RC.2", "release/1.6", "", RC_BASELINE_RECONCILED), "1.6.0-RC.3")
|
|
self.assertEqual(effective_version(RC_FROM_BRANCH, "1.6.0-SNAPSHOT", "release/1.6", "", RC_BASELINE_NONE), "1.6.0-RC.1")
|
|
|
|
def test_patch_bump_from_every_captured_format(self) -> None:
|
|
for declared in ("1.5.0", "1.5.0-SNAPSHOT", "1.5.0-RC.2", "1.5.0-new-abc1234-SNAPSHOT"):
|
|
self.assertEqual(effective_version(PATCH_BUMP, declared, "hotfix/1.5", ""), "1.5.1")
|
|
with self.assertRaises(PipelineError):
|
|
effective_version(PATCH_BUMP, "1.5.0", "release/1.5", "")
|
|
|
|
def test_stable_from_tag(self) -> None:
|
|
self.assertEqual(effective_version(STABLE_FROM_TAG, "1.5.0-RC.2", "", "v1.5.0"), "1.5.0")
|
|
for tag in ("", "v1.5", "v1.5.0-RC.1", "1.5.0"):
|
|
with self.assertRaises(PipelineError):
|
|
effective_version(STABLE_FROM_TAG, "1.5.0-RC.2", "", tag)
|
|
|
|
def test_feature_snapshot_is_always_refused_and_substitutes_no_name_or_hash(self) -> None:
|
|
"""No `publish` fallback name and no fixed 7-character hash: the originating feature branch and git's abbreviation are not carried inputs."""
|
|
for declared, branch, tag in (("1.5.0-SNAPSHOT", "", "publish-1"), ("1.5.0", "", "publish"), ("1.5.0-SNAPSHOT", "feature/x", "")):
|
|
with self.assertRaisesRegex(PipelineError, "originating feature branch"):
|
|
effective_version(FEATURE_SNAPSHOT, declared, branch, tag)
|
|
|
|
def test_unparseable_declared_version_is_an_error_for_every_rule(self) -> None:
|
|
for rule, branch, tag in ((DEVELOP_NEXT_MINOR, "develop", ""), (RC_FROM_BRANCH, "release/1.6", ""), (PATCH_BUMP, "hotfix/1.5", ""), (STABLE_FROM_TAG, "", "v1.5.0"), (FEATURE_SNAPSHOT, "", "publish")):
|
|
with self.assertRaises(PipelineError):
|
|
effective_version(rule, "undeclared", branch, tag)
|
|
self.assertEqual(version_parts("v2.3.4"), (2, 3, 4))
|
|
|
|
def test_unknown_rule_is_refused(self) -> None:
|
|
with self.assertRaises(PipelineError):
|
|
effective_version("latest-tag", "1.5.0", "main", "")
|
|
|
|
def test_property_conflict_is_refused(self) -> None:
|
|
self.assertEqual(with_version_property([["mvn", "verify"]], "revision", "1.6.0-RC.1"), [["mvn", "verify", "-Drevision=1.6.0-RC.1"]])
|
|
with self.assertRaises(PipelineError):
|
|
with_version_property([["mvn", "-Drevision=1", "verify"]], "revision", "1.6.0-RC.1")
|
|
|
|
|
|
class ConfigurationTest(unittest.TestCase):
|
|
def test_valid_rule_is_parsed_with_the_default_property(self) -> None:
|
|
channel = parse_project_configuration(_configuration()).channels[0]
|
|
self.assertEqual((channel.effective_version, channel.version_property), (RC_FROM_BRANCH, "revision"))
|
|
|
|
def test_a_verification_only_channel_without_publications_is_accepted(self) -> None:
|
|
"""The captured core profile ran the calculator for every publish kind; a repository that only verifies still declares the rule."""
|
|
self.assertEqual(parse_project_configuration(_configuration(publish=False)).channels[0].effective_version, RC_FROM_BRANCH)
|
|
|
|
def test_tag_and_branch_channels_take_their_rules(self) -> None:
|
|
for rule, trigger in ((STABLE_FROM_TAG, {"push_tags": ["v*.*.*"]}), (FEATURE_SNAPSHOT, {"push_tags": ["publish*"]}), (DEVELOP_NEXT_MINOR, {"push_branches": ["develop"]}), (PATCH_BUMP, {"push_branches": ["hotfix/**"]})):
|
|
with self.subTest(rule):
|
|
self.assertEqual(parse_project_configuration(_configuration(rule=rule, trigger=trigger, publish=False)).channels[0].effective_version, rule)
|
|
|
|
def test_absent_rule_changes_nothing(self) -> None:
|
|
configuration = _configuration()
|
|
cast("dict[str, dict[str, dict[str, object]]]", configuration["release"])["channels"]["candidate"].pop("effective_version")
|
|
self.assertEqual(parse_project_configuration(configuration).channels[0].effective_version, "")
|
|
|
|
def test_invalid_declarations_are_refused(self) -> None:
|
|
cases = {
|
|
"unknown rule": _configuration(rule="latest-tag"),
|
|
"bad property": _configuration(extra={"effective_version": {"rule": RC_FROM_BRANCH, "property": "a b"}}),
|
|
"missing rule": _configuration(extra={"effective_version": {"property": "revision"}}),
|
|
}
|
|
for label, configuration in cases.items():
|
|
with self.subTest(label), self.assertRaises(ConfigurationError):
|
|
parse_project_configuration(configuration)
|
|
|
|
|
|
class PlanApplicationTest(unittest.TestCase):
|
|
@override
|
|
def setUp(self) -> None:
|
|
self.project = parse_project_configuration(_configuration())
|
|
|
|
def test_push_applies_the_effective_version_to_every_command(self) -> None:
|
|
(target,) = _plan(self.project, push_facts("release/1.6", SOURCE_REVISION))
|
|
self.assertTrue(all(command[-1] == "-Drevision=1.6.0-RC.1" for command in target["commands"]))
|
|
|
|
def test_a_verification_only_component_gets_the_version_too(self) -> None:
|
|
project = parse_project_configuration(_configuration(publish=False))
|
|
(target,) = _plan(project, push_facts("release/1.6", SOURCE_REVISION))
|
|
self.assertEqual(target["commands"][0][-1], "-Drevision=1.6.0-RC.1")
|
|
|
|
def test_recorded_build_input_is_the_effective_version_and_digest_differs_from_the_declared_build(self) -> None:
|
|
(effective,) = _plan(self.project, push_facts("release/1.6", SOURCE_REVISION))
|
|
(declared,) = plan_targets(self.project, TEST_RUNTIME_CATALOG, EventFacts("pull_request", branch="feature/x", source_revision=SOURCE_REVISION))
|
|
# what packaging records: the same reader over the target's own commands, so the override is the version
|
|
recorded = compute_build_inputs(effective, SOURCE_TREE, _reader("1.5.0-SNAPSHOT")(effective))
|
|
self.assertEqual(recorded["version"], "1.6.0-RC.1")
|
|
self.assertNotEqual(recorded["config_digest"], compute_build_inputs(declared, SOURCE_TREE, "1.5.0-SNAPSHOT")["config_digest"])
|
|
|
|
def test_same_inputs_give_the_same_target(self) -> None:
|
|
facts = push_facts("release/1.6", SOURCE_REVISION)
|
|
self.assertEqual(_plan(self.project, facts), _plan(self.project, facts))
|
|
|
|
def test_a_different_declared_version_is_a_different_input(self) -> None:
|
|
facts = push_facts("release/1.6", SOURCE_REVISION)
|
|
(first,) = _plan(self.project, facts, "1.5.0-SNAPSHOT")
|
|
(second,) = _plan(self.project, facts, "1.6.0-RC.1")
|
|
self.assertEqual(second["commands"][0][-1], "-Drevision=1.6.0-RC.2")
|
|
self.assertNotEqual(first["commands"], second["commands"])
|
|
|
|
def test_pull_requests_plan_the_declared_property_and_unmatched_pushes_are_not_changed(self) -> None:
|
|
"""A pull request is the original verification of a published component, so it plans the explicit property with the declared version (see `test_unchanged_version_reuse`); a push no channel matches stays plain."""
|
|
(pull_request,) = _plan(self.project, EventFacts("pull_request", branch="release/1.6", source_revision=SOURCE_REVISION))
|
|
self.assertTrue(all(command[-1] == "-Drevision=1.5.0-SNAPSHOT" for command in pull_request["commands"]))
|
|
(unmatched,) = _plan(self.project, push_facts("develop", SOURCE_REVISION))
|
|
self.assertFalse(any(token.startswith("-Drevision=") for command in unmatched["commands"] for token in command))
|
|
|
|
def test_without_facts_nothing_is_applied(self) -> None:
|
|
(target,) = plan_targets(self.project, TEST_RUNTIME_CATALOG)
|
|
self.assertFalse(any(token.startswith("-Drevision=") for command in target["commands"] for token in command))
|
|
|
|
def test_a_rule_without_a_reader_is_refused(self) -> None:
|
|
with self.assertRaises(PipelineError):
|
|
plan_targets(self.project, TEST_RUNTIME_CATALOG, push_facts("release/1.6", SOURCE_REVISION))
|
|
|
|
def test_a_pom_that_does_not_resolve_through_the_property_is_refused(self) -> None:
|
|
with self.assertRaisesRegex(PipelineError, "resolves version"):
|
|
_plan(self.project, push_facts("release/1.6", SOURCE_REVISION), pom=LITERAL_POM)
|
|
|
|
def test_a_command_that_sets_the_property_is_refused(self) -> None:
|
|
configuration = _configuration()
|
|
component_mapping(configuration, "library")["build"] = {"tool": "maven", "arguments": ["-Drevision=1", "verify"]}
|
|
project = parse_project_configuration(configuration)
|
|
with self.assertRaises(PipelineError):
|
|
_plan(project, push_facts("release/1.6", SOURCE_REVISION))
|
|
|
|
def test_tag_rules_use_the_pushed_tag_and_commit(self) -> None:
|
|
stable = parse_project_configuration(_configuration(rule=STABLE_FROM_TAG, trigger={"push_tags": ["v*.*.*"]}, extra={"tag_policy": "stable"}))
|
|
(target,) = _plan(stable, tag_facts("v1.6.0", SOURCE_REVISION), "1.6.0-RC.2")
|
|
self.assertEqual(target["commands"][0][-1], "-Drevision=1.6.0")
|
|
feature = parse_project_configuration(_configuration(rule=FEATURE_SNAPSHOT, trigger={"push_tags": ["publish*"]}, publish=False))
|
|
with self.assertRaisesRegex(PipelineError, "originating feature branch"):
|
|
_plan(feature, tag_facts("publish-7", SOURCE_REVISION), "1.6.0-SNAPSHOT")
|
|
|
|
def test_the_reconciled_baseline_declaration_reaches_the_rule(self) -> None:
|
|
project = parse_project_configuration(_configuration(extra={"effective_version": {"rule": RC_FROM_BRANCH, "rc_baseline": "reconciled"}}))
|
|
self.assertEqual(project.channels[0].rc_baseline, "reconciled")
|
|
with self.assertRaisesRegex(PipelineError, "tagged first-parent RC counter"):
|
|
_plan(project, push_facts("release/1.6", SOURCE_REVISION), "1.6.0-SNAPSHOT")
|
|
(target,) = _plan(project, push_facts("release/1.6", SOURCE_REVISION), "1.6.0-RC.1")
|
|
self.assertEqual(target["commands"][0][-1], "-Drevision=1.6.0-RC.2")
|
|
|
|
def test_overlapping_channels_with_different_rules_are_refused(self) -> None:
|
|
configuration = _configuration()
|
|
channels = cast("dict[str, dict[str, object]]", cast("dict[str, object]", configuration["release"])["channels"])
|
|
channels["other"] = {"trigger": {"push_branches": ["release/**"]}, "effective_version": {"rule": PATCH_BUMP}}
|
|
with self.assertRaises(PipelineError):
|
|
_plan(parse_project_configuration(configuration), push_facts("release/1.6", SOURCE_REVISION))
|
|
|
|
|
|
class ProviderEquivalenceTest(unittest.TestCase):
|
|
def test_forge_style_reader_on_the_applied_target_agrees_with_the_build_record(self) -> None:
|
|
"""The publisher and tag planner recompute by reading the declared version of the APPLIED target; it must equal the version `prepare` applied."""
|
|
project = parse_project_configuration(_configuration())
|
|
facts = push_facts("release/1.6", SOURCE_REVISION)
|
|
(built,) = _plan(project, facts)
|
|
(recomputed,) = _plan(project, facts)
|
|
self.assertEqual(built, recomputed)
|
|
self.assertEqual(_reader("1.5.0-SNAPSHOT")(built), "1.6.0-RC.1")
|
|
|
|
|
|
class ReleaseActivityPredicateTest(unittest.TestCase):
|
|
def test_the_captured_merge_and_bump_messages_match(self) -> None:
|
|
for message in (
|
|
"Merge pull request 'Release 0.4' (#31) from release/0.4 into develop",
|
|
"Merge branch 'release/1.5.0' into develop",
|
|
"Merge branch 'release/1.5' back into develop\n\nbody",
|
|
"subject\n\nMerge release/2.0.1: done, back to develop",
|
|
"Bump version after release v1.5.0",
|
|
"Bump the version after release of v1.5.0 to 1.6.0-SNAPSHOT",
|
|
):
|
|
with self.subTest(message):
|
|
self.assertTrue(message_matches_release_activity(message))
|
|
|
|
def test_other_messages_do_not_match(self) -> None:
|
|
for message in (
|
|
"Merge branch 'release/next' into develop", # the patched matcher requires a numeric release/X.Y[.Z]
|
|
"Merge branch 'release/1.5x' into develop", # `x` continues the version token
|
|
"Merge branch 'release/1.5.0-RC.1' into main",
|
|
"Merge branch 'feature/x' into develop",
|
|
"Merge branch 'release/1.5'\ninto develop", # git greps one message line at a time
|
|
"Bump version to 1.6.0-SNAPSHOT",
|
|
"bump version after release v1", # case sensitive
|
|
"",
|
|
):
|
|
with self.subTest(message):
|
|
self.assertFalse(message_matches_release_activity(message))
|
|
|
|
def test_the_window_ends_at_the_build_start_and_includes_its_first_second(self) -> None:
|
|
start = 1_800_000_000
|
|
since = release_activity_since(start)
|
|
self.assertEqual(since, start - RELEASE_ACTIVITY_WINDOW_SECONDS)
|
|
self.assertEqual(RELEASE_ACTIVITY_WINDOW_SECONDS, 259_200)
|
|
merge = "Merge branch 'release/0.4' into develop"
|
|
self.assertTrue(has_release_activity([(since, merge)], since))
|
|
self.assertFalse(has_release_activity([(since - 1, merge)], since))
|
|
self.assertFalse(has_release_activity([(start, "ordinary change")], since))
|
|
|
|
|
|
REPOSITORY = "QRun-IO/example"
|
|
RUN_ID = 4242
|
|
PREPARE_NAME = f"Prepare plan [workflow {'a' * 40}]"
|
|
START = 1_800_000_000
|
|
DAY = 24 * 60 * 60
|
|
RELEASE_MERGE = "Merge pull request 'Release 0.4' (#31) from release/0.4 into develop"
|
|
SCOPE = RepositoryScope(REPOSITORY, 61, 1024 * 1024, 10, frozenset())
|
|
|
|
|
|
def _iso(epoch: int) -> str:
|
|
return datetime.fromtimestamp(epoch, UTC).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
class FakeForge:
|
|
"""A forge with runs (id -> prepare job start as epoch seconds or literal text; None leaves the field out), a newest-first commit history (a None time leaves the committer date out) and one `pom.xml`; every request is recorded."""
|
|
|
|
def __init__(self, starts: Mapping[int, int | str | None], commits: Sequence[tuple[int | None, str]] | None = None, pom: str | None = None, extra_jobs: Sequence[Mapping[str, object]] | None = None) -> None:
|
|
self.starts, self.extra_jobs = starts, list(extra_jobs or [])
|
|
self.commits = sorted(commits or [], key=lambda item: item[0] or 0, reverse=True)
|
|
self.pom = pom
|
|
self.requests: list[str] = []
|
|
|
|
def download(self, path: str, destination: Path, limit: int) -> None:
|
|
raise NotImplementedError(f"no artifact is served: {path} {destination} {limit}")
|
|
|
|
def get(self, path: str) -> object:
|
|
self.requests.append(path)
|
|
location, _, query = path.partition("?")
|
|
parameters = urllib.parse.parse_qs(query)
|
|
prefix = f"/repos/{REPOSITORY}/"
|
|
if location.startswith(f"{prefix}actions/runs/"):
|
|
run_id = int(location.removeprefix(f"{prefix}actions/runs/").split("/")[0])
|
|
if run_id not in self.starts:
|
|
raise self._missing(path)
|
|
if location.endswith("/jobs"):
|
|
started = self.starts[run_id]
|
|
prepare: dict[str, object] = {"name": PREPARE_NAME, "run_id": run_id, "head_sha": SOURCE_REVISION, "run_attempt": 1, **({"started_at": started if isinstance(started, str) else _iso(started)} if started is not None else {})}
|
|
jobs = [prepare, {"name": "Report", "run_id": run_id, "head_sha": SOURCE_REVISION, "run_attempt": 1}, *self.extra_jobs]
|
|
return {"total_count": len(jobs), "jobs": jobs}
|
|
return {"id": run_id, "head_sha": SOURCE_REVISION, "run_attempt": 1}
|
|
if location == f"{prefix}commits":
|
|
page, size = int(parameters["page"][0]), int(parameters["limit"][0])
|
|
return [{"sha": f"{index:040x}", "commit": {"message": message, "committer": {"date": _iso(when)} if when is not None else {}}} for index, (when, message) in enumerate(self.commits[(page - 1) * size : page * size])]
|
|
if location == f"{prefix}contents/pom.xml" and self.pom is not None:
|
|
data = self.pom.encode()
|
|
return {"type": "file", "encoding": "base64", "size": len(data), "content": base64.b64encode(data).decode()}
|
|
raise self._missing(path)
|
|
|
|
@staticmethod
|
|
def _missing(path: str) -> urllib.error.HTTPError:
|
|
return urllib.error.HTTPError(path, 404, "not found", Message(), io.BytesIO(b""))
|
|
|
|
|
|
class ServerClockBindingTest(unittest.TestCase):
|
|
def test_the_build_start_is_the_servers_prepare_job_start_floored_to_the_second(self) -> None:
|
|
self.assertEqual(prepare_job_started(FakeForge({RUN_ID: START}), SCOPE, RUN_ID), START)
|
|
self.assertEqual(prepare_job_started(FakeForge({RUN_ID: f"{_iso(START)[:-1]}.987654Z"}), SCOPE, RUN_ID), START)
|
|
|
|
def test_a_run_without_exactly_one_started_prepare_job_has_no_build_start(self) -> None:
|
|
with self.assertRaisesRegex(DeliveryError, "exactly one prepare job"):
|
|
prepare_job_started(FakeForge({RUN_ID: None}), SCOPE, RUN_ID)
|
|
second = {"name": PREPARE_NAME, "started_at": _iso(START)}
|
|
with self.assertRaisesRegex(DeliveryError, "exactly one prepare job"):
|
|
prepare_job_started(FakeForge({RUN_ID: START}, extra_jobs=[second]), SCOPE, RUN_ID)
|
|
with self.assertRaises(urllib.error.HTTPError):
|
|
prepare_job_started(FakeForge({}), SCOPE, RUN_ID)
|
|
|
|
def test_a_job_of_another_commit_is_refused(self) -> None:
|
|
foreign = {"name": "Verify x", "head_sha": "b" * 40, "run_id": RUN_ID, "run_attempt": 1}
|
|
with self.assertRaisesRegex(DeliveryError, "another run, commit or attempt"):
|
|
prepare_job_started(FakeForge({RUN_ID: START}, extra_jobs=[foreign]), SCOPE, RUN_ID)
|
|
|
|
def test_a_start_time_without_a_zone_is_not_a_trusted_timestamp(self) -> None:
|
|
with self.assertRaisesRegex(DeliveryError, "usable UTC timestamp"):
|
|
prepare_job_started(FakeForge({RUN_ID: "2027-01-15T08:00:00"}), SCOPE, RUN_ID)
|
|
with self.assertRaisesRegex(DeliveryError, "not a timestamp"):
|
|
prepare_job_started(FakeForge({RUN_ID: "yesterday"}), SCOPE, RUN_ID)
|
|
|
|
def test_the_answer_follows_the_runs_start_and_never_the_current_time(self) -> None:
|
|
"""The same history gives different answers for two runs, and the same run always gives the same one, so a later recomputation (publisher, tag, write-back) cannot drift."""
|
|
history: list[tuple[int | None, str]] = [(START - DAY, RELEASE_MERGE), (START - 5 * DAY, "ordinary")]
|
|
forge = FakeForge({1: START, 2: START + 3 * DAY + 1}, history)
|
|
self.assertTrue(release_activity_for_run(forge, SCOPE, SOURCE_REVISION, 1))
|
|
self.assertFalse(release_activity_for_run(forge, SCOPE, SOURCE_REVISION, 2))
|
|
self.assertTrue(release_activity_for_run(forge, SCOPE, SOURCE_REVISION, 1))
|
|
|
|
def test_commits_outside_the_window_or_not_matching_do_not_count(self) -> None:
|
|
older = FakeForge({RUN_ID: START}, [(START - 4 * DAY, RELEASE_MERGE), (START - DAY, "ordinary change")])
|
|
self.assertFalse(release_activity_for_run(older, SCOPE, SOURCE_REVISION, RUN_ID))
|
|
bump = FakeForge({RUN_ID: START}, [(START - 2 * DAY, "Bump version after release v0.4.0")])
|
|
self.assertTrue(release_activity_for_run(bump, SCOPE, SOURCE_REVISION, RUN_ID))
|
|
self.assertFalse(release_activity_for_run(FakeForge({RUN_ID: START}, []), SCOPE, SOURCE_REVISION, RUN_ID))
|
|
|
|
def test_a_match_on_a_later_page_is_found_and_reading_stops_at_the_first_page_outside_the_window(self) -> None:
|
|
recent = [(START - 100 - index, f"change {index}") for index in range(50)] # page 1: all inside the window, none matching
|
|
forge = FakeForge({RUN_ID: START}, [*recent, (START - 200, RELEASE_MERGE)])
|
|
self.assertTrue(release_activity_for_run(forge, SCOPE, SOURCE_REVISION, RUN_ID))
|
|
old = [(START - 10 * DAY - index, f"old {index}") for index in range(50)]
|
|
forge = FakeForge({RUN_ID: START}, [(START - 100, "recent"), *old, (START - 20 * DAY, RELEASE_MERGE)])
|
|
forge.requests.clear()
|
|
self.assertFalse(release_activity_for_run(forge, SCOPE, SOURCE_REVISION, RUN_ID))
|
|
pages = [request for request in forge.requests if "/commits?" in request]
|
|
self.assertEqual(len(pages), 2) # page 1 had one in-window commit, page 2 none: stop
|
|
|
|
def test_a_window_the_bound_cannot_cover_is_not_proven(self) -> None:
|
|
forge = FakeForge({RUN_ID: START}, [(START - 100 - index, f"change {index}") for index in range(100)])
|
|
with self.assertRaisesRegex(GitHistoryError, "not proven"):
|
|
ForgeGitHistory(forge, REPOSITORY, 2).has_release_activity(SOURCE_REVISION, release_activity_since(START))
|
|
|
|
def test_any_missing_fact_makes_the_version_unknown(self) -> None:
|
|
for label, forge in (("no run", FakeForge({})), ("no start", FakeForge({RUN_ID: None}))):
|
|
with self.subTest(label), self.assertRaises(ReleaseActivityError):
|
|
release_activity_for_run(forge, SCOPE, SOURCE_REVISION, RUN_ID)
|
|
with self.assertRaisesRegex(ReleaseActivityError, "committer date"):
|
|
release_activity_for_run(FakeForge({RUN_ID: START}, [(None, RELEASE_MERGE)]), SCOPE, SOURCE_REVISION, RUN_ID)
|
|
|
|
|
|
def _develop_project() -> ProjectConfiguration:
|
|
return parse_project_configuration(_configuration(rule=DEVELOP_NEXT_MINOR, trigger={"push_branches": ["develop"]}, extra={"versioning": "maven-snapshot"}))
|
|
|
|
|
|
class DevelopSnapshotFlowTest(unittest.TestCase):
|
|
"""The first real flow: declared `0.5.0-SNAPSHOT` on develop, in `prepare` and as the publisher recomputes it for the run it selects."""
|
|
|
|
def test_prepare_keeps_the_declared_snapshot_without_release_activity(self) -> None:
|
|
(target,) = _plan(_develop_project(), push_facts("develop", SOURCE_REVISION), "0.5.0-SNAPSHOT", release_activity=lambda: False)
|
|
self.assertTrue(all(command[-1] == "-Drevision=0.5.0-SNAPSHOT" for command in target["commands"]))
|
|
|
|
def test_prepare_bumps_after_recent_release_activity_and_the_inputs_differ(self) -> None:
|
|
facts = push_facts("develop", SOURCE_REVISION)
|
|
(kept,) = _plan(_develop_project(), facts, "0.5.0-SNAPSHOT", release_activity=lambda: False)
|
|
(bumped,) = _plan(_develop_project(), facts, "0.5.0-SNAPSHOT", release_activity=lambda: True)
|
|
self.assertEqual(bumped["commands"][0][-1], "-Drevision=0.6.0-SNAPSHOT")
|
|
kept_inputs = compute_build_inputs(kept, SOURCE_TREE, _reader("0.5.0-SNAPSHOT")(kept))
|
|
bumped_inputs = compute_build_inputs(bumped, SOURCE_TREE, _reader("0.5.0-SNAPSHOT")(bumped))
|
|
# reuse safety: another version is another config digest and another recorded version, so an earlier run's bytes cannot stand in for this one
|
|
self.assertEqual((kept_inputs["version"], bumped_inputs["version"]), ("0.5.0-SNAPSHOT", "0.6.0-SNAPSHOT"))
|
|
self.assertNotEqual(kept_inputs["config_digest"], bumped_inputs["config_digest"])
|
|
|
|
def test_prepare_without_the_input_fails_closed(self) -> None:
|
|
with self.assertRaises(ReleaseActivityError):
|
|
_plan(_develop_project(), push_facts("develop", SOURCE_REVISION), "0.5.0-SNAPSHOT")
|
|
|
|
def test_the_history_is_not_read_for_a_plan_that_does_not_need_it(self) -> None:
|
|
forge = FakeForge({RUN_ID: START})
|
|
_plan(_develop_project(), push_facts("develop", SOURCE_REVISION), "0.5.0-RC.1", release_activity=lambda: release_activity_for_run(forge, SCOPE, SOURCE_REVISION, RUN_ID))
|
|
_plan(_develop_project(), EventFacts("pull_request", branch="feature/x", source_revision=SOURCE_REVISION), "0.5.0-SNAPSHOT", release_activity=lambda: release_activity_for_run(forge, SCOPE, SOURCE_REVISION, RUN_ID))
|
|
self.assertEqual(forge.requests, [])
|
|
|
|
def test_the_publisher_recomputes_the_version_of_each_candidate_run_from_that_runs_start(self) -> None:
|
|
"""`prepare` of run 1 saw a release merge one day before its start and built 0.6.0-SNAPSHOT; run 2 started after the window closed and built 0.5.0-SNAPSHOT."""
|
|
forge = FakeForge({1: START, 2: START + 3 * DAY + 1}, [(START - DAY, RELEASE_MERGE)], POM.format(revision="0.5.0-SNAPSHOT"))
|
|
facts = push_facts("develop", SOURCE_REVISION)
|
|
project = _develop_project()
|
|
versions = {run: _plan_for_run(forge, SCOPE, TEST_RUNTIME_CATALOG, project, SOURCE_REVISION, facts, run)["library-linux-amd64"]["commands"][0][-1] for run in (1, 2)}
|
|
self.assertEqual(versions, {1: "-Drevision=0.6.0-SNAPSHOT", 2: "-Drevision=0.5.0-SNAPSHOT"})
|
|
# and each equals what `prepare` computes for the same run from the same facts
|
|
for run, expected in ((1, True), (2, False)):
|
|
(prepared,) = _plan(project, facts, "0.5.0-SNAPSHOT", release_activity=lambda run=run: release_activity_for_run(forge, SCOPE, SOURCE_REVISION, run))
|
|
self.assertEqual(prepared["commands"][0][-1], f"-Drevision={'0.6.0' if expected else '0.5.0'}-SNAPSHOT")
|
|
|
|
def test_a_run_whose_start_is_unknown_is_a_release_activity_error_for_that_run_only(self) -> None:
|
|
forge = FakeForge({1: START, 2: None}, [], POM.format(revision="0.5.0-SNAPSHOT"))
|
|
facts = push_facts("develop", SOURCE_REVISION)
|
|
project = _develop_project()
|
|
self.assertEqual(_plan_for_run(forge, SCOPE, TEST_RUNTIME_CATALOG, project, SOURCE_REVISION, facts, 1)["library-linux-amd64"]["commands"][0][-1], "-Drevision=0.5.0-SNAPSHOT")
|
|
with self.assertRaises(ReleaseActivityError):
|
|
_plan_for_run(forge, SCOPE, TEST_RUNTIME_CATALOG, project, SOURCE_REVISION, facts, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|