test_publishing.py (48465B)
1 #!/usr/bin/env python3 2 3 # This file is in the public domain. 4 5 import datetime 6 import importlib.machinery 7 import importlib.util 8 import subprocess 9 import sys 10 import unittest 11 from io import StringIO 12 from pathlib import Path 13 from tempfile import TemporaryDirectory 14 from types import SimpleNamespace 15 from unittest.mock import Mock, call, patch 16 17 ROOT = Path(__file__).parents[1] 18 sys.path.insert(0, str(ROOT)) 19 sys.path.insert(0, str(ROOT / "buildscripts")) 20 LOADER = importlib.machinery.SourceFileLoader( 21 "taler_pkg_publishing", str(ROOT / "taler-pkg") 22 ) 23 SPEC = importlib.util.spec_from_loader(LOADER.name, LOADER) 24 TALER_PKG = importlib.util.module_from_spec(SPEC) 25 LOADER.exec_module(TALER_PKG) 26 27 28 def result(stdout=""): 29 return subprocess.CompletedProcess([], 0, stdout=stdout) 30 31 32 def publication(prefix, distribution, source, kind): 33 return ( 34 f"Prefix: {prefix}\n" 35 f"Distribution: {distribution}\n" 36 "Architectures: amd64 arm64\n" 37 "Sources:\n" 38 f" main: {source} [{kind}]\n" 39 ) 40 41 42 class PublishingConfigTests(unittest.TestCase): 43 def test_distribution_mapping(self): 44 debian = TALER_PKG.publishing_config("debian-trixie") 45 ubuntu = TALER_PKG.publishing_config("ubuntu-noble") 46 47 self.assertEqual("apt/debian", debian.prefix) 48 self.assertEqual("taler-debian-trixie-testing", debian.testing_repo) 49 self.assertEqual("trixie-testing", debian.testing_distribution) 50 self.assertEqual("apt/ubuntu", ubuntu.prefix) 51 self.assertEqual("taler-ubuntu-noble-stable-initial", ubuntu.initial_snapshot) 52 with self.assertRaisesRegex(ValueError, "unsupported publishing distro"): 53 TALER_PKG.publishing_config("debian-bookworm") 54 55 def test_remote_command_quotes_every_remote_argument(self): 56 with patch.object(TALER_PKG.subprocess, "run", return_value=result()) as run: 57 TALER_PKG.remote_command( 58 ["command", "argument with spaces", "*.deb"], 59 capture_output=True, 60 tty=True, 61 ) 62 63 run.assert_called_once_with( 64 [ 65 "ssh", 66 "-t", 67 "taler-packaging@taler.net", 68 "command 'argument with spaces' '*.deb'", 69 ], 70 check=True, 71 capture_output=True, 72 text=True, 73 ) 74 75 def test_parses_publication_and_package_list(self): 76 parsed = TALER_PKG.parse_publication( 77 publication( 78 "apt/debian", 79 "trixie", 80 "taler-debian-trixie-testing-20260903T120000000000Z", 81 "snapshot", 82 ) 83 ) 84 packages = TALER_PKG.parse_package_list( 85 "Name: repo\nPackages:\n one_1_amd64\n two_2_all\n" 86 ) 87 88 self.assertEqual("apt/debian", parsed["prefix"]) 89 self.assertEqual( 90 ( 91 "taler-debian-trixie-testing-20260903T120000000000Z", 92 "snapshot", 93 ), 94 parsed["sources"]["main"], 95 ) 96 self.assertEqual(["one_1_amd64", "two_2_all"], packages) 97 98 def test_snapshot_names_are_utc_and_unique_to_microseconds(self): 99 pubcfg = TALER_PKG.publishing_config("debian-trixie") 100 now = datetime.datetime( 101 2026, 9, 3, 12, 34, 56, 123456, tzinfo=datetime.timezone.utc 102 ) 103 104 name = TALER_PKG.testing_snapshot_name(pubcfg, now) 105 106 self.assertEqual("taler-debian-trixie-testing-20260903T123456123456Z", name) 107 108 109 class InitializeTests(unittest.TestCase): 110 def test_initializes_all_missing_repositories_and_publications(self): 111 outputs = iter([result(), result(), result()]) 112 113 def aptly(*args, **kwargs): 114 if kwargs.get("capture_output"): 115 return next(outputs) 116 return result() 117 118 with ( 119 patch.object(TALER_PKG, "remote_aptly", side_effect=aptly) as remote, 120 patch.object( 121 TALER_PKG, 122 "testing_snapshot_name", 123 side_effect=lambda pubcfg: f"{pubcfg.testing_repo}-initial-test", 124 ), 125 patch("sys.stdout", new_callable=StringIO) as stdout, 126 ): 127 TALER_PKG.initialize(SimpleNamespace()) 128 129 calls = remote.call_args_list 130 self.assertIn( 131 call( 132 "repo", 133 "create", 134 "-distribution=trixie-testing", 135 "-component=main", 136 "taler-debian-trixie-testing", 137 ), 138 calls, 139 ) 140 self.assertIn( 141 call( 142 "publish", 143 "snapshot", 144 *TALER_PKG.publish_options("trixie-testing"), 145 "taler-debian-trixie-testing-initial-test", 146 "apt/debian", 147 tty=True, 148 ), 149 calls, 150 ) 151 self.assertIn( 152 call( 153 "snapshot", 154 "create", 155 "taler-ubuntu-noble-stable-initial", 156 "empty", 157 ), 158 calls, 159 ) 160 self.assertIn( 161 call( 162 "publish", 163 "snapshot", 164 *TALER_PKG.publish_options("noble"), 165 "taler-ubuntu-noble-stable-initial", 166 "apt/ubuntu", 167 tty=True, 168 ), 169 calls, 170 ) 171 self.assertFalse( 172 any( 173 item.args[:2] == ("repo", "create") 174 and item.args[-1].endswith("-stable") 175 for item in calls 176 ) 177 ) 178 self.assertIn("initialized", stdout.getvalue()) 179 for pubcfg in TALER_PKG.publishing_configs.values(): 180 self.assertIn( 181 call( 182 "snapshot", 183 "create", 184 f"{pubcfg.testing_repo}-initial-test", 185 "from", 186 "repo", 187 pubcfg.testing_repo, 188 ), 189 calls, 190 ) 191 192 def test_complete_initialization_is_a_noop(self): 193 repos = "\n".join( 194 pubcfg.testing_repo for pubcfg in TALER_PKG.publishing_configs.values() 195 ) 196 snapshots = "debian-snapshot\nubuntu-snapshot\n" 197 publications = ( 198 "apt/debian trixie-testing\n" 199 "apt/debian trixie\n" 200 "apt/ubuntu noble-testing\n" 201 "apt/ubuntu noble\n" 202 ) 203 204 def aptly(*args, **kwargs): 205 if args == ("repo", "list", "-raw"): 206 return result(repos) 207 if args == ("snapshot", "list", "-raw"): 208 return result(snapshots) 209 if args == ("publish", "list", "-raw"): 210 return result(publications) 211 if args[:2] == ("publish", "show"): 212 distribution, prefix = args[2:] 213 pubcfg = next( 214 item 215 for item in TALER_PKG.publishing_configs.values() 216 if item.prefix == prefix 217 ) 218 snapshot = ( 219 "debian-snapshot" 220 if pubcfg.vendor == "debian" 221 else "ubuntu-snapshot" 222 ) 223 return result(publication(prefix, distribution, snapshot, "snapshot")) 224 if args[:2] == ("snapshot", "show"): 225 pubcfg = ( 226 TALER_PKG.publishing_config("debian-trixie") 227 if args[2] == "debian-snapshot" 228 else TALER_PKG.publishing_config("ubuntu-noble") 229 ) 230 return result( 231 f"Description: Snapshot from local repo [{pubcfg.testing_repo}]\n" 232 ) 233 self.fail(f"unexpected aptly call: {args} {kwargs}") 234 235 with ( 236 patch.object(TALER_PKG, "remote_aptly", side_effect=aptly) as remote, 237 patch("sys.stdout", new_callable=StringIO) as stdout, 238 ): 239 TALER_PKG.initialize(SimpleNamespace()) 240 241 self.assertFalse( 242 any(not item.kwargs.get("capture_output") for item in remote.mock_calls) 243 ) 244 self.assertIn("already initialized", stdout.getvalue()) 245 246 def test_reuses_initial_snapshots_when_only_stable_publish_is_missing(self): 247 repos = "\n".join( 248 pubcfg.testing_repo for pubcfg in TALER_PKG.publishing_configs.values() 249 ) 250 snapshots = "\n".join( 251 pubcfg.initial_snapshot for pubcfg in TALER_PKG.publishing_configs.values() 252 ) 253 publications = "apt/debian trixie-testing\napt/ubuntu noble-testing\n" 254 255 def aptly(*args, **kwargs): 256 if args == ("repo", "list", "-raw"): 257 return result(repos) 258 if args == ("snapshot", "list", "-raw"): 259 return result(snapshots) 260 if args == ("publish", "list", "-raw"): 261 return result(publications) 262 if args[:2] == ("publish", "show"): 263 distribution, prefix = args[2:] 264 pubcfg = next( 265 item 266 for item in TALER_PKG.publishing_configs.values() 267 if item.prefix == prefix 268 ) 269 return result( 270 publication( 271 prefix, 272 distribution, 273 pubcfg.testing_repo + "-snapshot", 274 "snapshot", 275 ) 276 ) 277 if args[:2] == ("snapshot", "show"): 278 if args[2].endswith("-snapshot"): 279 repo = args[2].removesuffix("-snapshot") 280 return result(f"Description: Snapshot from local repo [{repo}]\n") 281 return result( 282 "Description: Snapshot from local repo [legacy-stable-repo]\n" 283 "Packages:\n" 284 ) 285 return result() 286 287 with patch.object(TALER_PKG, "remote_aptly", side_effect=aptly) as remote: 288 TALER_PKG.initialize(SimpleNamespace()) 289 290 mutations = [ 291 item for item in remote.mock_calls if not item.kwargs.get("capture_output") 292 ] 293 self.assertEqual(2, len(mutations)) 294 self.assertTrue( 295 all(item.args[:2] == ("publish", "snapshot") for item in mutations) 296 ) 297 298 def test_rejects_legacy_testing_publication(self): 299 pubcfg = TALER_PKG.publishing_config("debian-trixie") 300 wrong = publication( 301 pubcfg.prefix, 302 pubcfg.testing_distribution, 303 pubcfg.testing_repo, 304 "local", 305 ) 306 with self.assertRaisesRegex(ValueError, "convert it to a snapshot publication"): 307 TALER_PKG.validate_publication( 308 pubcfg, TALER_PKG.parse_publication(wrong), stable=False 309 ) 310 311 def test_accepts_empty_initial_snapshot_for_stable_publication(self): 312 pubcfg = TALER_PKG.publishing_config("debian-trixie") 313 initial = publication( 314 pubcfg.prefix, 315 pubcfg.codename, 316 pubcfg.initial_snapshot, 317 "snapshot", 318 ) 319 320 with ( 321 patch.object(TALER_PKG, "snapshot_packages", return_value=[]), 322 patch.object(TALER_PKG, "snapshot_origin") as snapshot_origin, 323 ): 324 TALER_PKG.validate_publication( 325 pubcfg, TALER_PKG.parse_publication(initial), stable=True 326 ) 327 328 snapshot_origin.assert_not_called() 329 330 def test_rejects_nonempty_initial_snapshot(self): 331 pubcfg = TALER_PKG.publishing_config("debian-trixie") 332 initial = publication( 333 pubcfg.prefix, 334 pubcfg.codename, 335 pubcfg.initial_snapshot, 336 "snapshot", 337 ) 338 339 with ( 340 patch.object( 341 TALER_PKG, "snapshot_packages", return_value=["package_1_amd64"] 342 ), 343 self.assertRaisesRegex(ValueError, "initial snapshot .* is not empty"), 344 ): 345 TALER_PKG.validate_publication( 346 pubcfg, TALER_PKG.parse_publication(initial), stable=True 347 ) 348 349 def test_rejects_snapshot_with_wrong_origin_in_either_channel(self): 350 pubcfg = TALER_PKG.publishing_config("debian-trixie") 351 for stable in (False, True): 352 with self.subTest(stable=stable): 353 details = publication( 354 pubcfg.prefix, 355 pubcfg.codename if stable else pubcfg.testing_distribution, 356 "unexpected-snapshot", 357 "snapshot", 358 ) 359 with ( 360 patch.object( 361 TALER_PKG, "snapshot_origin", return_value="unexpected-repo" 362 ), 363 self.assertRaisesRegex(ValueError, "unexpected source"), 364 ): 365 TALER_PKG.validate_publication( 366 pubcfg, TALER_PKG.parse_publication(details), stable=stable 367 ) 368 369 def test_initialization_rejects_legacy_testing_without_replacing_it(self): 370 pubcfg = TALER_PKG.publishing_config("debian-trixie") 371 with ( 372 patch.object(TALER_PKG, "aptly_list", return_value={pubcfg.testing_repo}), 373 patch.object( 374 TALER_PKG, 375 "published_repositories", 376 return_value={(pubcfg.prefix, pubcfg.testing_distribution)}, 377 ), 378 patch.object( 379 TALER_PKG, 380 "get_publication", 381 return_value=TALER_PKG.parse_publication( 382 publication( 383 pubcfg.prefix, 384 pubcfg.testing_distribution, 385 pubcfg.testing_repo, 386 "local", 387 ) 388 ), 389 ), 390 patch.object(TALER_PKG, "remote_aptly") as aptly, 391 self.assertRaisesRegex(ValueError, "convert it to a snapshot publication"), 392 ): 393 TALER_PKG.initialize(SimpleNamespace()) 394 aptly.assert_not_called() 395 396 397 class PublishingInventoryTests(unittest.TestCase): 398 def test_reads_each_source_once_and_ignores_other_pools(self): 399 pubcfg = TALER_PKG.publishing_config("debian-trixie") 400 publications = { 401 ("apt/debian", "trixie-testing"): { 402 "main": (pubcfg.testing_repo, "local"), 403 "contrib": ("ignored-repo", "local"), 404 }, 405 ("apt/debian", "trixie"): {"main": ("stable", "snapshot")}, 406 ("apt/debian", "stable-alias"): {"main": ("stable", "snapshot")}, 407 ("apt/debian", "legacy"): {"main": ("legacy", "local")}, 408 ("apt/debian", "extras"): {"contrib": ("ignored-repo", "local")}, 409 ("apt/ubuntu", "noble"): {"main": ("ignored-snapshot", "snapshot")}, 410 ("filesystem:other:apt/debian", "trixie"): { 411 "main": ("ignored-snapshot", "snapshot") 412 }, 413 } 414 repos = { 415 pubcfg.testing_repo: ["testing_1_amd64", "common_1_all"], 416 "legacy": ["legacy_2_arm64"], 417 } 418 419 def get_publication(distribution, prefix): 420 return { 421 "prefix": prefix, 422 "distribution": distribution, 423 "sources": publications[prefix, distribution], 424 } 425 426 with ( 427 patch.object( 428 TALER_PKG, "published_repositories", return_value=set(publications) 429 ), 430 patch.object( 431 TALER_PKG, "get_publication", side_effect=get_publication 432 ) as show, 433 patch.object( 434 TALER_PKG, "repo_packages", side_effect=repos.__getitem__ 435 ) as repo, 436 patch.object( 437 TALER_PKG, 438 "snapshot_packages", 439 return_value=["stable_2_amd64", "common_1_all"], 440 ) as snapshot, 441 ): 442 packages = TALER_PKG.publishing_packages(pubcfg) 443 444 self.assertEqual( 445 ["common_1_all", "legacy_2_arm64", "stable_2_amd64", "testing_1_amd64"], 446 packages, 447 ) 448 self.assertCountEqual( 449 [call(pubcfg.testing_repo), call("legacy")], repo.call_args_list 450 ) 451 snapshot.assert_called_once_with("stable") 452 self.assertEqual(5, show.call_count) 453 self.assertTrue( 454 all(item.args[1] == "apt/debian" for item in show.call_args_list) 455 ) 456 457 def test_inventory_errors_prevent_uploads_and_cleanup(self): 458 pubcfg = TALER_PKG.publishing_config("debian-trixie") 459 for failure in ("list", "show", "repo", "snapshot", "kind", "details"): 460 with self.subTest(failure=failure): 461 error = subprocess.CalledProcessError(1, ["aptly", failure]) 462 details = TALER_PKG.parse_publication( 463 publication("apt/debian", "trixie", "stable", "snapshot") 464 ) 465 if failure == "kind": 466 details["sources"]["main"] = ("stable", "unsupported") 467 elif failure == "details": 468 details = {"sources": {}} 469 with ( 470 patch.object(TALER_PKG, "published_snapshot"), 471 patch.object( 472 TALER_PKG, 473 "published_repositories", 474 return_value={("apt/debian", "trixie")}, 475 side_effect=error if failure == "list" else None, 476 ), 477 patch.object( 478 TALER_PKG, 479 "get_publication", 480 return_value=details, 481 side_effect=error if failure == "show" else None, 482 ), 483 patch.object( 484 TALER_PKG, 485 "repo_packages", 486 return_value=[], 487 side_effect=error if failure == "repo" else None, 488 ), 489 patch.object( 490 TALER_PKG, 491 "snapshot_packages", 492 return_value=[], 493 side_effect=error if failure == "snapshot" else None, 494 ), 495 patch.object(TALER_PKG, "remote_command") as command, 496 patch.object(TALER_PKG, "remote_aptly") as aptly, 497 patch.object(TALER_PKG.subprocess, "run") as run, 498 patch.object(TALER_PKG, "cleanup_uploads") as cleanup, 499 self.assertRaises((ValueError, subprocess.CalledProcessError)), 500 ): 501 TALER_PKG.publish(SimpleNamespace(distro=pubcfg.distro, dry=False)) 502 503 command.assert_not_called() 504 aptly.assert_not_called() 505 run.assert_not_called() 506 cleanup.assert_not_called() 507 508 509 class PublishTests(unittest.TestCase): 510 def setUp(self): 511 self.publications = self.enterContext( 512 patch.object(TALER_PKG, "published_repositories", return_value=set()) 513 ) 514 self.published_snapshot = self.enterContext( 515 patch.object( 516 TALER_PKG, 517 "published_snapshot", 518 return_value=("previous", {"amd64", "arm64"}), 519 ) 520 ) 521 self.snapshot = "taler-testing-20260906T120000000000Z" 522 self.enterContext( 523 patch.object(TALER_PKG, "testing_snapshot_name", return_value=self.snapshot) 524 ) 525 526 def test_uploads_only_newer_packages_and_updates_testing(self): 527 local_packages = [ 528 "merchant_2.0_amd64.deb", 529 "merchant-dbgsym_2.0_amd64.ddeb", 530 "exchange_1.0_amd64.deb", 531 "manual_1.0_all.deb", 532 ] 533 server_packages = [ 534 "merchant_1.0_amd64", 535 "merchant_3.0_arm64", 536 "merchant-dbgsym_1.0_amd64", 537 "exchange_2.0_amd64", 538 "manual_1.0_all", 539 ] 540 with ( 541 patch.object( 542 TALER_PKG, "current_package_files", return_value=local_packages 543 ), 544 patch.object(TALER_PKG, "repo_packages", return_value=server_packages), 545 patch.object(TALER_PKG, "remote_command") as remote_command, 546 patch.object(TALER_PKG, "remote_aptly") as remote_aptly, 547 patch.object(TALER_PKG.subprocess, "run") as run, 548 patch.object(TALER_PKG, "cleanup_uploads") as cleanup, 549 ): 550 operations = Mock() 551 operations.attach_mock(remote_command, "command") 552 operations.attach_mock(run, "upload") 553 operations.attach_mock(remote_aptly, "aptly") 554 operations.attach_mock(cleanup, "cleanup") 555 TALER_PKG.publish(SimpleNamespace(distro="debian-trixie", dry=False)) 556 557 rsync = run.call_args.args[0] 558 self.assertIn(Path("packages/debian-trixie/merchant_2.0_amd64.deb"), rsync) 559 self.assertIn( 560 Path("packages/debian-trixie/merchant-dbgsym_2.0_amd64.ddeb"), rsync 561 ) 562 self.assertNotIn(Path("packages/debian-trixie/exchange_1.0_amd64.deb"), rsync) 563 self.assertNotIn(Path("packages/debian-trixie/manual_1.0_all.deb"), rsync) 564 remote_command.assert_called_once_with( 565 ["mkdir", "-p", "/home/taler-packaging/debian-trixie"] 566 ) 567 self.assertEqual( 568 ["command", "upload", "aptly", "aptly", "aptly", "cleanup"], 569 [item[0] for item in operations.mock_calls], 570 ) 571 cleanup.assert_called_once_with("/home/taler-packaging/debian-trixie") 572 remote_aptly.assert_has_calls( 573 [ 574 call( 575 "repo", 576 "add", 577 "taler-debian-trixie-testing", 578 "/home/taler-packaging/debian-trixie/merchant_2.0_amd64.deb", 579 "/home/taler-packaging/debian-trixie/merchant-dbgsym_2.0_amd64.ddeb", 580 ), 581 call( 582 "snapshot", 583 "create", 584 self.snapshot, 585 "from", 586 "repo", 587 "taler-debian-trixie-testing", 588 ), 589 call( 590 "publish", 591 "switch", 592 "trixie-testing", 593 "apt/debian", 594 self.snapshot, 595 tty=True, 596 ), 597 ] 598 ) 599 self.assertFalse( 600 any( 601 call.args[0:2] == ("repo", "remove") for call in remote_aptly.mock_calls 602 ) 603 ) 604 605 def test_skips_equal_and_older_versions_in_published_sources(self): 606 local_packages = [ 607 "merchant_2.0_amd64.deb", 608 "merchant-dbgsym_2.0_amd64.ddeb", 609 "exchange_1.0_amd64.deb", 610 "manual_1.0_all.deb", 611 ] 612 server_packages = [ 613 "merchant_2.0_amd64", 614 "merchant-dbgsym_2.0_amd64", 615 "exchange_2.0_amd64", 616 "manual_1.0_all", 617 ] 618 for kind in ("local", "snapshot"): 619 for distribution in ("trixie", "legacy"): 620 for dry in (False, True): 621 with self.subTest(kind=kind, distribution=distribution, dry=dry): 622 self.publications.return_value = {("apt/debian", distribution)} 623 details = TALER_PKG.parse_publication( 624 publication("apt/debian", distribution, "published", kind) 625 ) 626 with ( 627 patch.object( 628 TALER_PKG, "get_publication", return_value=details 629 ), 630 patch.object( 631 TALER_PKG, 632 "repo_packages", 633 side_effect=lambda name: ( 634 server_packages if name == "published" else [] 635 ), 636 ), 637 patch.object( 638 TALER_PKG, 639 "snapshot_packages", 640 return_value=server_packages, 641 ), 642 patch.object( 643 TALER_PKG, 644 "current_package_files", 645 return_value=local_packages, 646 ), 647 patch.object(TALER_PKG, "remote_command") as command, 648 patch.object(TALER_PKG, "remote_aptly") as aptly, 649 patch.object(TALER_PKG.subprocess, "run") as run, 650 patch.object(TALER_PKG, "cleanup_uploads") as cleanup, 651 patch("sys.stdout", new_callable=StringIO) as stdout, 652 ): 653 TALER_PKG.publish( 654 SimpleNamespace(distro="debian-trixie", dry=dry) 655 ) 656 657 command.assert_not_called() 658 run.assert_not_called() 659 for package in server_packages: 660 self.assertIn(f"server has {package}", stdout.getvalue()) 661 if dry: 662 aptly.assert_not_called() 663 cleanup.assert_not_called() 664 else: 665 self.assertEqual( 666 [ 667 call( 668 "snapshot", 669 "create", 670 self.snapshot, 671 "from", 672 "repo", 673 "taler-debian-trixie-testing", 674 ), 675 call( 676 "publish", 677 "switch", 678 "trixie-testing", 679 "apt/debian", 680 self.snapshot, 681 tty=True, 682 ), 683 ], 684 aptly.call_args_list, 685 ) 686 cleanup.assert_called_once_with( 687 "/home/taler-packaging/debian-trixie" 688 ) 689 690 def test_newer_versions_and_other_architectures_are_uploaded(self): 691 self.publications.return_value = {("apt/debian", "trixie")} 692 local_packages = [ 693 "merchant_2.0-1+trixie_amd64.deb", 694 "merchant_2.0-0+trixie_arm64.deb", 695 ] 696 with ( 697 patch.object( 698 TALER_PKG, 699 "get_publication", 700 return_value=TALER_PKG.parse_publication( 701 publication("apt/debian", "trixie", "stable", "snapshot") 702 ), 703 ), 704 patch.object(TALER_PKG, "repo_packages", return_value=[]), 705 patch.object( 706 TALER_PKG, 707 "snapshot_packages", 708 return_value=["merchant_2.0-0+trixie_amd64"], 709 ), 710 patch.object( 711 TALER_PKG, "current_package_files", return_value=local_packages 712 ), 713 patch.object(TALER_PKG, "remote_command"), 714 patch.object(TALER_PKG, "remote_aptly") as aptly, 715 patch.object(TALER_PKG.subprocess, "run") as run, 716 patch.object(TALER_PKG, "cleanup_uploads"), 717 ): 718 TALER_PKG.publish(SimpleNamespace(distro="debian-trixie", dry=False)) 719 720 for filename in local_packages: 721 self.assertIn( 722 Path("packages/debian-trixie") / filename, run.call_args.args[0] 723 ) 724 self.assertEqual( 725 call( 726 "repo", 727 "add", 728 "taler-debian-trixie-testing", 729 *( 730 f"/home/taler-packaging/debian-trixie/{name}" 731 for name in local_packages 732 ), 733 ), 734 aptly.call_args_list[0], 735 ) 736 737 def test_rejects_non_package_or_path_artifacts(self): 738 for filename in ( 739 "merchant.changes", 740 "merchant.deb", 741 "../merchant_2.0_amd64.deb", 742 ): 743 with self.subTest(filename=filename): 744 with self.assertRaisesRegex(ValueError, "invalid package"): 745 TALER_PKG.package_file_identity(filename) 746 747 def test_dry_publish_never_mutates(self): 748 with ( 749 patch.object( 750 TALER_PKG, 751 "current_package_files", 752 return_value=["merchant_2.0_amd64.deb"], 753 ), 754 patch.object(TALER_PKG, "repo_packages", return_value=[]), 755 patch.object(TALER_PKG, "remote_command") as remote_command, 756 patch.object(TALER_PKG, "remote_aptly") as remote_aptly, 757 patch.object(TALER_PKG.subprocess, "run") as run, 758 patch.object(TALER_PKG, "cleanup_uploads") as cleanup, 759 ): 760 TALER_PKG.publish(SimpleNamespace(distro="debian-trixie", dry=True)) 761 762 remote_command.assert_not_called() 763 remote_aptly.assert_not_called() 764 run.assert_not_called() 765 cleanup.assert_not_called() 766 767 def test_empty_publish_still_refreshes_testing_metadata(self): 768 with ( 769 patch.object(TALER_PKG, "current_package_files", return_value=[]), 770 patch.object(TALER_PKG, "repo_packages", return_value=[]), 771 patch.object(TALER_PKG, "remote_command") as remote_command, 772 patch.object(TALER_PKG, "remote_aptly") as remote_aptly, 773 patch.object(TALER_PKG.subprocess, "run") as run, 774 patch.object(TALER_PKG, "cleanup_uploads") as cleanup, 775 ): 776 TALER_PKG.publish(SimpleNamespace(distro="ubuntu-noble", dry=False)) 777 778 remote_command.assert_not_called() 779 run.assert_not_called() 780 self.assertEqual( 781 [ 782 call( 783 "snapshot", 784 "create", 785 self.snapshot, 786 "from", 787 "repo", 788 "taler-ubuntu-noble-testing", 789 ), 790 call( 791 "publish", 792 "switch", 793 "noble-testing", 794 "apt/ubuntu", 795 self.snapshot, 796 tty=True, 797 ), 798 ], 799 remote_aptly.call_args_list, 800 ) 801 cleanup.assert_called_once_with("/home/taler-packaging/ubuntu-noble") 802 803 def test_failed_upload_import_snapshot_or_switch_keeps_staged_files(self): 804 for phase in ("upload", "import", "snapshot", "switch"): 805 with self.subTest(phase=phase): 806 error = subprocess.CalledProcessError(1, [phase]) 807 results = [result()] * max( 808 0, ("upload", "import", "snapshot", "switch").index(phase) - 1 809 ) + [error] 810 811 with ( 812 patch.object(TALER_PKG, "repo_packages", return_value=[]), 813 patch.object( 814 TALER_PKG, 815 "current_package_files", 816 return_value=["merchant_2.0_amd64.deb"], 817 ), 818 patch.object(TALER_PKG, "remote_command") as command, 819 patch.object( 820 TALER_PKG, 821 "remote_aptly", 822 side_effect=results, 823 ), 824 patch.object( 825 TALER_PKG.subprocess, 826 "run", 827 side_effect=error if phase == "upload" else None, 828 ), 829 patch.object(TALER_PKG, "cleanup_uploads") as cleanup, 830 self.assertRaises(subprocess.CalledProcessError), 831 ): 832 TALER_PKG.publish( 833 SimpleNamespace(distro="debian-trixie", dry=False) 834 ) 835 836 command.assert_called_once_with( 837 ["mkdir", "-p", "/home/taler-packaging/debian-trixie"] 838 ) 839 cleanup.assert_not_called() 840 841 def test_cleanup_failure_reports_that_publication_succeeded(self): 842 error = subprocess.CalledProcessError(1, ["cleanup"]) 843 with ( 844 patch.object(TALER_PKG, "repo_packages", return_value=[]), 845 patch.object(TALER_PKG, "current_package_files", return_value=[]), 846 patch.object(TALER_PKG, "remote_aptly") as aptly, 847 patch.object(TALER_PKG, "cleanup_uploads", side_effect=error), 848 self.assertRaisesRegex( 849 RuntimeError, 850 "Publishing trixie-testing succeeded, but cleanup .* failed", 851 ), 852 ): 853 TALER_PKG.publish(SimpleNamespace(distro="debian-trixie", dry=False)) 854 855 self.assertEqual( 856 [ 857 call( 858 "snapshot", 859 "create", 860 self.snapshot, 861 "from", 862 "repo", 863 "taler-debian-trixie-testing", 864 ), 865 call( 866 "publish", 867 "switch", 868 "trixie-testing", 869 "apt/debian", 870 self.snapshot, 871 tty=True, 872 ), 873 ], 874 aptly.call_args_list, 875 ) 876 877 878 class UploadCleanupTests(unittest.TestCase): 879 def test_cleanup_only_removes_top_level_regular_package_files(self): 880 with TemporaryDirectory() as tmp: 881 staging = Path(tmp) / "upload dir 'quoted'" 882 staging.mkdir() 883 for filename in ( 884 "package.deb", 885 "debug.ddeb", 886 "notes.txt", 887 "package.changes", 888 ): 889 (staging / filename).write_text("test data") 890 nested = staging / "nested" 891 nested.mkdir() 892 (nested / "package.deb").write_text("nested data") 893 outside = Path(tmp) / "outside.deb" 894 outside.write_text("outside data") 895 (staging / "link.deb").symlink_to(outside) 896 with patch.object( 897 TALER_PKG, 898 "remote_command", 899 side_effect=lambda command: subprocess.run(command, check=True), 900 ): 901 TALER_PKG.cleanup_uploads(str(staging)) 902 903 self.assertEqual( 904 {"notes.txt", "package.changes", "nested", "link.deb"}, 905 {item.name for item in staging.iterdir()}, 906 ) 907 self.assertEqual("nested data", (nested / "package.deb").read_text()) 908 self.assertEqual("outside data", outside.read_text()) 909 910 def test_missing_upload_directory_is_a_noop(self): 911 with TemporaryDirectory() as tmp: 912 staging = Path(tmp) / "missing" 913 with patch.object( 914 TALER_PKG, 915 "remote_command", 916 side_effect=lambda command: subprocess.run(command, check=True), 917 ): 918 TALER_PKG.cleanup_uploads(str(staging)) 919 self.assertFalse(staging.exists()) 920 921 922 class PromoteTests(unittest.TestCase): 923 def test_dry_promote_shows_stable_to_testing_diff(self): 924 snapshots = { 925 "stable-snapshot": ["common_1_all", "removed_1_amd64"], 926 "testing-snapshot": ["common_1_all", "new_2_amd64"], 927 } 928 with ( 929 patch.object( 930 TALER_PKG, 931 "published_snapshot", 932 side_effect=lambda pubcfg, stable: ( 933 "stable-snapshot" if stable else "testing-snapshot", 934 {"amd64", "arm64"}, 935 ), 936 ), 937 patch.object( 938 TALER_PKG, 939 "snapshot_packages", 940 side_effect=snapshots.__getitem__, 941 ), 942 patch.object( 943 TALER_PKG, 944 "repo_packages", 945 return_value=["common_1_all", "new_3_amd64", "pending_1_all"], 946 ) as repo, 947 patch.object(TALER_PKG, "remote_aptly") as remote_aptly, 948 patch("sys.stdout", new_callable=StringIO) as stdout, 949 ): 950 TALER_PKG.promote(SimpleNamespace(distro="debian-trixie", dry=True)) 951 952 self.assertEqual("- removed_1_amd64\n+ new_2_amd64\n", stdout.getvalue()) 953 remote_aptly.assert_not_called() 954 repo.assert_not_called() 955 956 def test_promote_reuses_the_published_testing_snapshot(self): 957 snapshot = "taler-ubuntu-noble-testing-20260903T123456123456Z" 958 with ( 959 patch.object( 960 TALER_PKG, 961 "published_snapshot", 962 return_value=(snapshot, {"amd64", "arm64"}), 963 ) as published, 964 patch.object( 965 TALER_PKG, "repo_packages", return_value=["pending_2_all"] 966 ) as repo, 967 patch.object(TALER_PKG, "remote_aptly") as remote, 968 ): 969 TALER_PKG.promote(SimpleNamespace(distro="ubuntu-noble", dry=False)) 970 971 published.assert_called_once_with( 972 TALER_PKG.publishing_config("ubuntu-noble"), stable=False 973 ) 974 remote.assert_called_once_with( 975 "publish", "switch", "noble", "apt/ubuntu", snapshot, tty=True 976 ) 977 repo.assert_not_called() 978 979 980 class ShowPublishedTests(unittest.TestCase): 981 def test_all_target_forms_list_only_the_selected_published_snapshot(self): 982 for pubcfg in TALER_PKG.publishing_configs.values(): 983 for suffix in ("", "-stable", "-testing"): 984 with self.subTest(distro=pubcfg.distro, suffix=suffix): 985 distribution = ( 986 pubcfg.testing_distribution 987 if suffix == "-testing" 988 else pubcfg.codename 989 ) 990 snapshot = f"published-{distribution}" 991 992 def aptly( 993 *args, 994 distribution=distribution, 995 pubcfg=pubcfg, 996 snapshot=snapshot, 997 **kwargs, 998 ): 999 self.assertEqual({"capture_output": True}, kwargs) 1000 if args == ("publish", "show", distribution, pubcfg.prefix): 1001 return result( 1002 publication( 1003 pubcfg.prefix, distribution, snapshot, "snapshot" 1004 ) 1005 ) 1006 if args == ("snapshot", "show", "-with-packages", snapshot): 1007 return result( 1008 "Packages:\n two_2_all\n one_1_arm64\n one_1_amd64\n" 1009 " two_2_all\n excluded_1_i386\n source_1_source\n" 1010 ) 1011 self.fail(f"unexpected aptly call: {args}") 1012 1013 with ( 1014 patch.object( 1015 TALER_PKG, "remote_aptly", side_effect=aptly 1016 ) as remote, 1017 patch.object( 1018 TALER_PKG, "repo_packages", return_value=["pending_3_all"] 1019 ) as repo, 1020 patch.object( 1021 sys, 1022 "argv", 1023 ["taler-pkg", "show-published", pubcfg.distro + suffix], 1024 ), 1025 patch("sys.stdout", new_callable=StringIO) as stdout, 1026 ): 1027 TALER_PKG.main() 1028 1029 self.assertEqual( 1030 "one_1_amd64\none_1_arm64\ntwo_2_all\n", stdout.getvalue() 1031 ) 1032 self.assertEqual(2, remote.call_count) 1033 repo.assert_not_called() 1034 1035 def test_latest_uses_debian_versions_per_package_and_architecture(self): 1036 packages = [ 1037 "one_1.9_amd64", 1038 "one_1.10~rc1_amd64", 1039 "one_1.10_amd64", 1040 "one_1.10-2_amd64", 1041 "one_1.10-10_amd64", 1042 "one_1.8_arm64", 1043 "one_1.9_arm64", 1044 "one_1.7_all", 1045 "one_1.8_all", 1046 "two_9.0_all", 1047 "two_1:1.0_all", 1048 "two_1:1.0_all", 1049 "excluded_99_i386", 1050 "source_99_source", 1051 ] 1052 newest = [ 1053 "one_1.10-10_amd64", 1054 "one_1.9_arm64", 1055 "one_1.8_all", 1056 "two_1:1.0_all", 1057 ] 1058 for distro in TALER_PKG.publishing_configs: 1059 for suffix in ("", "-stable", "-testing"): 1060 for latest in (False, True): 1061 with ( 1062 self.subTest(distro=distro, suffix=suffix, latest=latest), 1063 patch.object( 1064 TALER_PKG, 1065 "published_snapshot", 1066 return_value=("selected", {"amd64", "arm64"}), 1067 ) as published, 1068 patch.object( 1069 TALER_PKG, "snapshot_packages", return_value=packages 1070 ) as snapshot, 1071 patch.object( 1072 sys, 1073 "argv", 1074 ["taler-pkg", "show-published", distro + suffix] 1075 + (["--latest"] if latest else []), 1076 ), 1077 patch("sys.stdout", new_callable=StringIO) as stdout, 1078 ): 1079 TALER_PKG.main() 1080 expected = newest if latest else packages[:-2] 1081 self.assertEqual( 1082 "".join(f"{package}\n" for package in sorted(set(expected))), 1083 stdout.getvalue(), 1084 ) 1085 published.assert_called_once_with( 1086 TALER_PKG.publishing_config(distro), 1087 stable=suffix != "-testing", 1088 ) 1089 snapshot.assert_called_once_with("selected") 1090 1091 def test_unsupported_targets_fail_before_contacting_server(self): 1092 for target in ( 1093 "debian-bookworm", 1094 "debian-trixie-stable-testing", 1095 "debian-trixie-experimental", 1096 ): 1097 with ( 1098 self.subTest(target=target), 1099 patch.object(sys, "argv", ["taler-pkg", "show-published", target]), 1100 patch.object(TALER_PKG, "remote_aptly") as remote, 1101 patch("sys.stderr", new_callable=StringIO), 1102 self.assertRaises(SystemExit) as error, 1103 ): 1104 TALER_PKG.main() 1105 self.assertEqual(2, error.exception.code) 1106 remote.assert_not_called() 1107 1108 def test_empty_published_snapshot_prints_nothing(self): 1109 with ( 1110 patch.object( 1111 TALER_PKG, "published_snapshot", return_value=("empty", {"amd64"}) 1112 ), 1113 patch.object(TALER_PKG, "snapshot_packages", return_value=[]), 1114 patch("sys.stdout", new_callable=StringIO) as stdout, 1115 ): 1116 TALER_PKG.show_published( 1117 SimpleNamespace(distro="ubuntu-noble-testing", latest=True) 1118 ) 1119 self.assertEqual("", stdout.getvalue()) 1120 1121 def test_unpublished_architectures_are_excluded(self): 1122 with ( 1123 patch.object( 1124 TALER_PKG, "published_snapshot", return_value=("selected", {"arm64"}) 1125 ), 1126 patch.object( 1127 TALER_PKG, 1128 "snapshot_packages", 1129 return_value=["one_1_amd64", "one_1_arm64", "two_2_all"], 1130 ), 1131 patch("sys.stdout", new_callable=StringIO) as stdout, 1132 ): 1133 TALER_PKG.show_published( 1134 SimpleNamespace(distro="debian-trixie-testing", latest=True) 1135 ) 1136 self.assertEqual("one_1_arm64\ntwo_2_all\n", stdout.getvalue()) 1137 1138 def test_failed_lookups_do_not_print_a_partial_listing(self): 1139 for phase in ("publication", "snapshot"): 1140 with self.subTest(phase=phase): 1141 error = subprocess.CalledProcessError(1, ["aptly"]) 1142 details = result( 1143 publication("apt/debian", "trixie-testing", "published", "snapshot") 1144 ) 1145 with ( 1146 patch.object( 1147 TALER_PKG, 1148 "remote_aptly", 1149 side_effect=[error] 1150 if phase == "publication" 1151 else [details, error], 1152 ), 1153 patch("sys.stdout", new_callable=StringIO) as stdout, 1154 self.assertRaises(subprocess.CalledProcessError), 1155 ): 1156 TALER_PKG.show_published( 1157 SimpleNamespace(distro="debian-trixie-testing", latest=True) 1158 ) 1159 self.assertEqual("", stdout.getvalue()) 1160 1161 1162 class PublishedSnapshotTests(unittest.TestCase): 1163 def test_legacy_testing_is_rejected_before_listing_uploading_or_promoting(self): 1164 for operation in ( 1165 TALER_PKG.show_published, 1166 TALER_PKG.publish, 1167 TALER_PKG.promote, 1168 ): 1169 with self.subTest(operation=operation.__name__): 1170 target = ( 1171 "debian-trixie-testing" 1172 if operation == TALER_PKG.show_published 1173 else "debian-trixie" 1174 ) 1175 with ( 1176 patch.object( 1177 TALER_PKG, 1178 "remote_aptly", 1179 return_value=result( 1180 publication( 1181 "apt/debian", 1182 "trixie-testing", 1183 "taler-debian-trixie-testing", 1184 "local", 1185 ) 1186 ), 1187 ) as remote, 1188 patch.object(TALER_PKG, "remote_command") as command, 1189 patch.object(TALER_PKG, "publishing_packages") as inventory, 1190 patch.object(TALER_PKG, "cleanup_uploads") as cleanup, 1191 self.assertRaisesRegex( 1192 ValueError, "convert it to a snapshot publication" 1193 ), 1194 ): 1195 operation(SimpleNamespace(distro=target, dry=False)) 1196 remote.assert_called_once_with( 1197 "publish", 1198 "show", 1199 "trixie-testing", 1200 "apt/debian", 1201 capture_output=True, 1202 ) 1203 command.assert_not_called() 1204 inventory.assert_not_called() 1205 cleanup.assert_not_called() 1206 1207 def test_invalid_publication_details_are_rejected(self): 1208 for field, value in ( 1209 ("prefix", "apt/ubuntu"), 1210 ("distribution", "trixie"), 1211 ("architectures", set()), 1212 ("sources", {}), 1213 ): 1214 with self.subTest(field=field): 1215 details = TALER_PKG.parse_publication( 1216 publication("apt/debian", "trixie-testing", "selected", "snapshot") 1217 ) 1218 details[field] = value 1219 with ( 1220 patch.object(TALER_PKG, "get_publication", return_value=details), 1221 self.assertRaises(ValueError), 1222 ): 1223 TALER_PKG.published_snapshot( 1224 TALER_PKG.publishing_config("debian-trixie"), stable=False 1225 ) 1226 1227 1228 if __name__ == "__main__": 1229 unittest.main()