ci: separate manual release publishing
CI / lint (pull_request) Successful in 12s
CI / typecheck (pull_request) Successful in 10s
CI / test (pull_request) Successful in 17s
CI / build (pull_request) Successful in 11s
CI / package (pull_request) Successful in 13s
CI / integration (pull_request) Successful in 7m19s
CI / lint (pull_request) Successful in 12s
CI / typecheck (pull_request) Successful in 10s
CI / test (pull_request) Successful in 17s
CI / build (pull_request) Successful in 11s
CI / package (pull_request) Successful in 13s
CI / integration (pull_request) Successful in 7m19s
This commit is contained in:
+63
-15
@@ -63,9 +63,13 @@ def release_bundle(directory: Path) -> list[Path]:
|
||||
|
||||
class FakeGitea(http.server.BaseHTTPRequestHandler):
|
||||
assets: dict[str, bytes] = {}
|
||||
release_exists = True
|
||||
tag_revision: str | None = REVISION
|
||||
api_authorizations: list[str | None] = []
|
||||
public_authorizations: list[str | None] = []
|
||||
posts = 0
|
||||
asset_posts = 0
|
||||
release_posts = 0
|
||||
release_request: dict[str, object] | None = None
|
||||
|
||||
def log_message(self, _format: str, *_args: object) -> None:
|
||||
pass
|
||||
@@ -100,9 +104,15 @@ class FakeGitea(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
self.api_authorizations.append(self.headers.get("Authorization"))
|
||||
if "/releases/tags/" in self.path:
|
||||
self.answer(200, {"id": 7})
|
||||
if self.release_exists:
|
||||
self.answer(200, {"id": 7, "tag_name": f"v{VERSION}"})
|
||||
else:
|
||||
self.answer(404, {"message": "not found"})
|
||||
elif "/tags/" in self.path:
|
||||
self.answer(200, {"commit": {"sha": REVISION}})
|
||||
if self.tag_revision is not None:
|
||||
self.answer(200, {"commit": {"sha": self.tag_revision}})
|
||||
else:
|
||||
self.answer(404, {"message": "not found"})
|
||||
elif self.path.endswith("/releases/7/assets"):
|
||||
self.answer(200, self.asset_list())
|
||||
else:
|
||||
@@ -110,10 +120,19 @@ class FakeGitea(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler's API
|
||||
self.api_authorizations.append(self.headers.get("Authorization"))
|
||||
self.__class__.posts += 1
|
||||
name = urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query)["name"][0]
|
||||
length = int(self.headers["Content-Length"])
|
||||
body = self.rfile.read(length)
|
||||
if self.path.endswith("/releases"):
|
||||
request = json.loads(body)
|
||||
self.__class__.release_posts += 1
|
||||
self.__class__.release_request = request
|
||||
self.__class__.release_exists = True
|
||||
self.__class__.tag_revision = request["target_commitish"]
|
||||
self.answer(201, {"id": 7, "tag_name": request["tag_name"]})
|
||||
return
|
||||
|
||||
self.__class__.asset_posts += 1
|
||||
name = urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query)["name"][0]
|
||||
boundary = self.headers["Content-Type"].split("boundary=", 1)[1].encode()
|
||||
payload = body.split(b"\r\n\r\n", 1)[1].rsplit(b"\r\n--" + boundary + b"--", 1)[0]
|
||||
self.assets[name] = payload
|
||||
@@ -121,11 +140,16 @@ class FakeGitea(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def fake_gitea(initial: dict[str, bytes] | None = None):
|
||||
def fake_gitea(initial: dict[str, bytes] | None = None, *, release_exists: bool = True,
|
||||
tag_revision: str | None = REVISION):
|
||||
FakeGitea.assets = dict(initial or {})
|
||||
FakeGitea.release_exists = release_exists
|
||||
FakeGitea.tag_revision = tag_revision
|
||||
FakeGitea.api_authorizations = []
|
||||
FakeGitea.public_authorizations = []
|
||||
FakeGitea.posts = 0
|
||||
FakeGitea.asset_posts = 0
|
||||
FakeGitea.release_posts = 0
|
||||
FakeGitea.release_request = None
|
||||
with socketserver.TCPServer(("127.0.0.1", 0), FakeGitea) as server:
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
@@ -174,13 +198,25 @@ class ReleaseAssetsTest(unittest.TestCase):
|
||||
with self.assertRaisesRegex(release_assets.Refused, "does not match"):
|
||||
release_assets.expected_paths(directory, f"v{VERSION}", REVISION)
|
||||
|
||||
def test_uploads_each_selected_byte_and_reads_it_back_anonymously(self) -> None:
|
||||
def test_creates_release_uploads_each_byte_and_reads_it_back_anonymously(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
directory = Path(temporary)
|
||||
expected = release_bundle(directory)
|
||||
with fake_gitea():
|
||||
release_assets.upload(directory, f"v{VERSION}", REVISION)
|
||||
self.assertEqual(FakeGitea.posts, 6)
|
||||
with fake_gitea(release_exists=False, tag_revision=None):
|
||||
release_assets.publish(directory, f"v{VERSION}", REVISION)
|
||||
self.assertEqual(FakeGitea.release_posts, 1)
|
||||
self.assertEqual(FakeGitea.release_request, {
|
||||
"tag_name": f"v{VERSION}",
|
||||
"target_commitish": REVISION,
|
||||
"name": f"Ekklesia Signage v{VERSION}",
|
||||
"body": (
|
||||
"Published by the manual Publish Signage Release workflow after package, "
|
||||
"external integration, and corresponding-source verification."
|
||||
),
|
||||
"draft": False,
|
||||
"prerelease": True,
|
||||
})
|
||||
self.assertEqual(FakeGitea.asset_posts, 6)
|
||||
self.assertEqual(FakeGitea.assets, {path.name: path.read_bytes() for path in expected})
|
||||
self.assertTrue(all(value == "token test-token" for value in FakeGitea.api_authorizations))
|
||||
self.assertEqual(FakeGitea.public_authorizations, [None] * 6)
|
||||
@@ -191,14 +227,26 @@ class ReleaseAssetsTest(unittest.TestCase):
|
||||
expected = release_bundle(directory)
|
||||
initial = {path.name: path.read_bytes() for path in expected}
|
||||
with fake_gitea(initial):
|
||||
release_assets.upload(directory, f"v{VERSION}", REVISION)
|
||||
self.assertEqual(FakeGitea.posts, 0)
|
||||
release_assets.publish(directory, f"v{VERSION}", REVISION)
|
||||
self.assertEqual(FakeGitea.release_posts, 0)
|
||||
self.assertEqual(FakeGitea.asset_posts, 0)
|
||||
|
||||
conflicting = {expected[1].name: b"different published bytes"}
|
||||
with fake_gitea(conflicting):
|
||||
with self.assertRaisesRegex(release_assets.Refused, "public .* expected"):
|
||||
release_assets.upload(directory, f"v{VERSION}", REVISION)
|
||||
self.assertEqual(FakeGitea.posts, 0)
|
||||
release_assets.publish(directory, f"v{VERSION}", REVISION)
|
||||
self.assertEqual(FakeGitea.release_posts, 0)
|
||||
self.assertEqual(FakeGitea.asset_posts, 0)
|
||||
|
||||
def test_refuses_existing_tag_at_another_revision_before_mutation(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
directory = Path(temporary)
|
||||
release_bundle(directory)
|
||||
with fake_gitea(release_exists=False, tag_revision="b" * 40):
|
||||
with self.assertRaisesRegex(release_assets.Refused, "existing tag"):
|
||||
release_assets.publish(directory, f"v{VERSION}", REVISION)
|
||||
self.assertEqual(FakeGitea.release_posts, 0)
|
||||
self.assertEqual(FakeGitea.asset_posts, 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user