Use the BuiltinSSHAgentSocketProvider type in the tests
Marco Ricci

Marco Ricci commited on 2026-06-15 22:27:21
Zeige 3 geänderte Dateien mit 68 Einfügungen und 28 Löschungen.


Again, instead of magic strings floating around in the test suite, use
the already existing enum instead.
... ...
@@ -515,7 +515,7 @@ The standard registry of agent spawning functions, annotated as
515 515
 a `pytest` parameter set.  (In particular, this includes the conditional
516 516
 skip marks.)  Used by some test fixtures.
517 517
 """
518
-for key, handler in spawn_handlers.items():
518
+for instance, handler in spawn_handlers.items():
519 519
     marks = [
520 520
         pytest.mark.skipif(
521 521
             not is_agent_permitted(handler.instance.agent_type),
... ...
@@ -537,7 +537,10 @@ for key, handler in spawn_handlers.items():
537 537
     if handler.instance.agent_type != data.KnownSSHAgentType.StubbedSSHAgent:
538 538
         marks.append(pytest_machinery.non_isolated_agent_use)
539 539
     # Some agents require UNIX domain socket support.
540
-    if key in {"unix-pageant", "openssh"}:
540
+    if instance in {
541
+        data.KnownSSHAgentInstance.unix_pageant,
542
+        data.KnownSSHAgentInstance.openssh,
543
+    }:
541 544
         marks.append(
542 545
             pytest.mark.skipif(
543 546
                 not hasattr(socket, "AF_UNIX"),
... ...
@@ -545,14 +548,19 @@ for key, handler in spawn_handlers.items():
545 548
             )
546 549
         )
547 550
     # Others require Windows named pipe support.
548
-    if key in {"pageant", "openssh-on-windows"}:
551
+    if instance in {
552
+        data.KnownSSHAgentInstance.pageant,
553
+        data.KnownSSHAgentInstance.openssh_on_windows,
554
+    }:
549 555
         marks.append(
550 556
             pytest.mark.skipif(
551 557
                 not hasattr(ctypes, "WinDLL"),
552 558
                 reason="No Python or system support for Windows named pipes.",
553 559
             )
554 560
         )
555
-    spawn_handlers_params.append(pytest.param(handler, id=key, marks=marks))
561
+    spawn_handlers_params.append(
562
+        pytest.param(handler, id=instance, marks=marks)
563
+    )
556 564
 
557 565
 
558 566
 Popen = TypeVar("Popen", bound=subprocess.Popen)
... ...
@@ -692,10 +700,12 @@ def spawn_named_agent(  # noqa: C901
692 700
             monkeypatch.setattr(
693 701
                 ssh_agent.SSHAgentClient,
694 702
                 "SOCKET_PROVIDERS",
695
-                ["stub_agent_with_address_and_deterministic_dsa"]
703
+                [
704
+                    data.KnownSSHAgentInstance.stub_agent_with_address_and_deterministic_dsa
705
+                ]
696 706
                 if "stub_agent_with_address_and_deterministic_dsa"
697 707
                 in agent_name
698
-                else ["stub_agent_with_address"],
708
+                else [data.KnownSSHAgentInstance.stub_agent_with_address],
699 709
             )
700 710
             client = exit_stack.enter_context(
701 711
                 ssh_agent.SSHAgentClient.ensure_agent_subcontext(
... ...
@@ -801,7 +811,9 @@ def running_ssh_agent(  # pragma: no cover [external]
801 811
                 monkeypatch.setattr(
802 812
                     ssh_agent.SSHAgentClient,
803 813
                     "SOCKET_PROVIDERS",
804
-                    ["stub_agent_with_address"],
814
+                    [
815
+                        _types.BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS
816
+                    ],
805 817
                 )
806 818
                 monkeypatch.setenv(
807 819
                     "SSH_AUTH_SOCK",
... ...
@@ -386,7 +386,9 @@ class AgentProtocolResponseQueue:
386 386
 # ------------
387 387
 
388 388
 
389
-@socketprovider.SocketProvider.register("stub_agent")
389
+@socketprovider.SocketProvider.register(
390
+    _types.BuiltinSSHAgentSocketProvider.STUB_AGENT
391
+)
390 392
 class StubbedSSHAgentSocket:
391 393
     """A stubbed SSH agent presenting an [`_types.SSHAgentSocket`][]."""
392 394
 
... ...
@@ -678,7 +680,9 @@ class StubbedSSHAgentSocket:
678 680
 # ----------------
679 681
 
680 682
 
681
-@socketprovider.SocketProvider.register("stub_agent_with_address")
683
+@socketprovider.SocketProvider.register(
684
+    _types.BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS
685
+)
682 686
 class StubbedSSHAgentSocketWithAddress(StubbedSSHAgentSocket):
683 687
     """A [`StubbedSSHAgentSocket`][] requiring a specific address."""
684 688
 
... ...
@@ -724,7 +728,7 @@ class StubbedSSHAgentSocketWithAddress(StubbedSSHAgentSocket):
724 728
 
725 729
 
726 730
 @socketprovider.SocketProvider.register(
727
-    "stub_agent_with_address_and_deterministic_dsa"
731
+    _types.BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA
728 732
 )
729 733
 class StubbedSSHAgentSocketWithAddressAndDeterministicDSA(
730 734
     StubbedSSHAgentSocketWithAddress
... ...
@@ -690,12 +690,20 @@ class TestSSHAgentSocketProviderRegistry:
690 690
         resolve = socketprovider.SocketProvider.resolve
691 691
         lookup = socketprovider.SocketProvider.lookup
692 692
         with pytest.MonkeyPatch.context() as monkeypatch:
693
-            monkeypatch.setitem(registry, "stub_agent", None)
694
-            assert callable(lookup("native"))
695
-            assert callable(resolve("native"))
696
-            assert lookup("stub_agent") is None
693
+            monkeypatch.setitem(
694
+                registry, _types.BuiltinSSHAgentSocketProvider.STUB_AGENT, None
695
+            )
696
+            assert callable(
697
+                lookup(_types.BuiltinSSHAgentSocketProvider.NATIVE)
698
+            )
699
+            assert callable(
700
+                resolve(_types.BuiltinSSHAgentSocketProvider.NATIVE)
701
+            )
702
+            assert (
703
+                lookup(_types.BuiltinSSHAgentSocketProvider.STUB_AGENT) is None
704
+            )
697 705
             with pytest.raises(NotImplementedError):
698
-                resolve("stub_agent")
706
+                resolve(_types.BuiltinSSHAgentSocketProvider.STUB_AGENT)
699 707
 
700 708
     @Parametrize.RESOLVE_CHAINS
701 709
     def test_resolve_chains(
... ...
@@ -708,15 +716,17 @@ class TestSSHAgentSocketProviderRegistry:
708 716
         resolve = socketprovider.SocketProvider.resolve
709 717
         lookup = socketprovider.SocketProvider.lookup
710 718
         try:
711
-            implementation = resolve("native")
719
+            implementation = resolve(
720
+                _types.BuiltinSSHAgentSocketProvider.NATIVE
721
+            )
712 722
         except NotImplementedError:  # pragma: no cover
713 723
             pytest.fail("Native SSH agent socket provider is unavailable?!")
714 724
         # TODO(the-13th-letter): Rewrite using structural pattern matching.
715 725
         # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
716
-        target = (
726
+        target: _types.SSHAgentSocketProvider | str | None = (
717 727
             None
718 728
             if terminal == "unimplemented"
719
-            else "native"
729
+            else _types.BuiltinSSHAgentSocketProvider.NATIVE
720 730
             if terminal == "alias"
721 731
             else implementation
722 732
         )
... ...
@@ -767,16 +777,18 @@ class TestSSHAgentSocketProviderRegistry:
767 777
         resolve = socketprovider.SocketProvider.resolve
768 778
         lookup = socketprovider.SocketProvider.lookup
769 779
         try:
770
-            implementation = resolve("native")
780
+            implementation = resolve(
781
+                _types.BuiltinSSHAgentSocketProvider.NATIVE
782
+            )
771 783
         except NotImplementedError:  # pragma: no cover
772 784
             hypothesis.note(f"{registry = }")
773 785
             pytest.fail("Native SSH agent socket provider is unavailable?!")
774 786
         # TODO(the-13th-letter): Rewrite using structural pattern matching.
775 787
         # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
776
-        target = (
788
+        target: _types.SSHAgentSocketProvider | str | None = (
777 789
             None
778 790
             if terminal == "unimplemented"
779
-            else "native"
791
+            else _types.BuiltinSSHAgentSocketProvider.NATIVE
780 792
             if terminal == "alias"
781 793
             else implementation
782 794
         )
... ...
@@ -878,15 +890,21 @@ class TestSSHAgentSocketProviderRegistry:
878 890
     ) -> None:
879 891
         """Resolving a non-existant entry fails."""
880 892
         new_registry = {
881
-            "posix": socketprovider.SocketProvider.resolve("posix"),
882
-            "windows": socketprovider.SocketProvider.resolve("windows"),
893
+            _types.BuiltinSSHAgentSocketProvider.POSIX: socketprovider.SocketProvider.resolve(
894
+                _types.BuiltinSSHAgentSocketProvider.POSIX
895
+            ),
896
+            _types.BuiltinSSHAgentSocketProvider.WINDOWS: socketprovider.SocketProvider.resolve(
897
+                _types.BuiltinSSHAgentSocketProvider.WINDOWS
898
+            ),
883 899
         }
884 900
         with pytest.MonkeyPatch.context() as monkeypatch:
885 901
             monkeypatch.setattr(
886 902
                 socketprovider.SocketProvider, "registry", new_registry
887 903
             )
888 904
             with pytest.raises(socketprovider.NoSuchProviderError):
889
-                socketprovider.SocketProvider.resolve("native")
905
+                socketprovider.SocketProvider.resolve(
906
+                    _types.BuiltinSSHAgentSocketProvider.NATIVE
907
+                )
890 908
 
891 909
     def test_register_new_entry(
892 910
         self,
... ...
@@ -898,8 +916,12 @@ class TestSSHAgentSocketProviderRegistry:
898 916
 
899 917
         names = ["spam", "ham", "eggs", "parrot"]
900 918
         new_registry = {
901
-            "posix": socketprovider.SocketProvider.resolve("posix"),
902
-            "windows": socketprovider.SocketProvider.resolve("windows"),
919
+            _types.BuiltinSSHAgentSocketProvider.POSIX: socketprovider.SocketProvider.resolve(
920
+                _types.BuiltinSSHAgentSocketProvider.POSIX
921
+            ),
922
+            _types.BuiltinSSHAgentSocketProvider.WINDOWS: socketprovider.SocketProvider.resolve(
923
+                _types.BuiltinSSHAgentSocketProvider.WINDOWS
924
+            ),
903 925
         }
904 926
         with pytest.MonkeyPatch.context() as monkeypatch:
905 927
             monkeypatch.setattr(
... ...
@@ -1483,7 +1505,9 @@ class TestSSHAgentSocketProviderFailures:
1483 1505
     ) -> None:
1484 1506
         """For `posix`, fail if the running agent cannot be located."""
1485 1507
         del skip_if_no_af_unix_support
1486
-        provider = socketprovider.SocketProvider.resolve("posix")
1508
+        provider = socketprovider.SocketProvider.resolve(
1509
+            _types.BuiltinSSHAgentSocketProvider.POSIX
1510
+        )
1487 1511
         with pytest.MonkeyPatch.context() as monkeypatch:
1488 1512
             monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
1489 1513
             with pytest.raises(
... ...
@@ -1500,7 +1524,7 @@ class TestSSHAgentSocketProviderFailures:
1500 1524
         """For `ssh_auth_sock_on_windows`, fail for bad named pipe names."""
1501 1525
         del skip_if_no_ctypes_windll_support
1502 1526
         provider = socketprovider.SocketProvider.resolve(
1503
-            "ssh_auth_sock_on_windows"
1527
+            _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS
1504 1528
         )
1505 1529
         with pytest.MonkeyPatch.context() as monkeypatch:
1506 1530
             monkeypatch.setenv("SSH_AUTH_SOCK", named_pipe_name)
1507 1531