test genie on 3.10, better pyfakefs workaround for pyats issue

requirements-genie.txt

@@ -1,2 +1,2 @@
-genie>=20.2 ; sys_platform != "win32" and python_version < "3.10"
-pyats>=20.2 ; sys_platform != "win32" and python_version < "3.10"
+genie>=20.2 ; sys_platform != "win32" and python_version < "3.11"
+pyats>=20.2 ; sys_platform != "win32" and python_version < "3.11"

tests/conftest.py

@@ -9,12 +9,21 @@
     cisco_nxos_clean_response,
     juniper_junos_clean_response,
 )
+from pyfakefs.fake_filesystem_unittest import Patcher
 
 import scrapli
 
 TEST_DATA_PATH = f"{Path(scrapli.__file__).parents[1]}/tests/test_data"
 
 
+@pytest.fixture
+def fs_():
+    # replaces "fs" (pyfakefs) with patched one that does *not* use cache -- this is because
+    # pyats does something fucky with modules that breaks pyfakefs.
+    with Patcher(use_cache=False) as patcher:
+        yield patcher.fs
+
+
 @pytest.fixture(scope="session")
 def test_data_path():
     """Fixture to provide path to test data files"""

tests/unit/channel/test_async_channel.py

@@ -38,9 +38,8 @@ async def test_channel_lock_context_manager_no_channel_lock(async_transport_no_a
         assert True
 
 
-async def test_channel_read(fs, caplog, monkeypatch, async_transport_no_abc):
+async def test_channel_read(fs_, caplog, monkeypatch, async_transport_no_abc):
     # fs needed to mock filesystem for asserting log location
-    _ = fs
     caplog.set_level(logging.DEBUG, logger="scrapli.channel")
 
     channel_read_called = False

tests/unit/channel/test_base_channel.py

@@ -47,8 +47,8 @@ def test_channel_auth_properties(test_data, base_channel):
     assert getattr(base_channel, property_name) == compiled_new_value
 
 
-def test_channel_log_append(fs, base_transport_no_abc):
-    fs.create_file(
+def test_channel_log_append(fs_, base_transport_no_abc):
+    fs_.create_file(
         "scrapli_channel.log",
         contents="APPEND TO ME PLEASE!",
     )
@@ -67,18 +67,16 @@ def test_channel_log_invalid_mode(base_transport_no_abc):
         BaseChannelArgs(channel_log=True, channel_log_mode="not valid")
 
 
-def test_channel_log(fs, base_transport_no_abc):
+def test_channel_log(fs_, base_transport_no_abc):
     # fs needed to mock filesystem for asserting log location
-    _ = fs
     base_channel_args = BaseChannelArgs(channel_log=True)
     chan = BaseChannel(transport=base_transport_no_abc, base_channel_args=base_channel_args)
     chan.open()
     assert Path("/scrapli_channel.log").is_file()
 
 
-def test_channel_log_user_defined(fs, base_transport_no_abc):
+def test_channel_log_user_defined(fs_, base_transport_no_abc):
     # fs needed to mock filesystem for asserting log location
-    _ = fs
     base_channel_args = BaseChannelArgs(channel_log="/log.log")
     chan = BaseChannel(transport=base_transport_no_abc, base_channel_args=base_channel_args)
     chan.open()

tests/unit/channel/test_sync_channel.py

@@ -32,9 +32,8 @@ def test_channel_lock_context_manager_no_channel_lock(base_transport_no_abc):
         assert True
 
 
-def test_channel_read(fs, caplog, monkeypatch, sync_transport_no_abc):
+def test_channel_read(fs_, caplog, monkeypatch, sync_transport_no_abc):
     # fs needed to mock filesystem for asserting log location
-    _ = fs
     caplog.set_level(logging.DEBUG, logger="scrapli.channel")
 
     channel_read_called = False

tests/unit/driver/base/test_base_base_driver.py

@@ -90,8 +90,8 @@ def test_setup_ssh_file_args_telnet_transport(caplog, base_driver, test_data):
     assert logging.DEBUG == log_record.levelno
 
 
-def test_update_ssh_args_from_ssh_config(fs, real_ssh_config_file_path, base_driver):
-    fs.add_real_file(source_path=real_ssh_config_file_path, target_path="ssh_config")
+def test_update_ssh_args_from_ssh_config(fs_, real_ssh_config_file_path, base_driver):
+    fs_.add_real_file(source_path=real_ssh_config_file_path, target_path="ssh_config")
     base_driver.ssh_config_file = "ssh_config"
     base_driver.host = "1.2.3.4"
     base_driver.port = 0
@@ -119,16 +119,14 @@ def test_update_ssh_args_from_ssh_config(fs, real_ssh_config_file_path, base_dri
     ),
     ids=("true", "unresolvable_path"),
 )
-def test_setup_ssh_file_args_resolved(fs, base_driver, test_data):
+def test_setup_ssh_file_args_resolved(fs_, base_driver, test_data):
     """
     Assert we handle ssh config/known hosts inputs properly
 
     This does *not* need to test resolution as there is a test for that, this is just making sure
     that if given a non False bool or a string we properly try to resolve the ssh files
     """
     # using fakefs to ensure we dont resolve user/system config files
-    _ = fs
-
     ssh_config_file_input, ssh_known_hosts_file_input = test_data
 
     resolved_ssh_config_file, resolved_ssh_known_hosts_file = base_driver._setup_ssh_file_args(
@@ -240,11 +238,11 @@ def _import_module(_):
     ],
     ids=("auto_etc", "auto_user", "manual_location", "no_config"),
 )
-def test_resolve_ssh_config(fs, real_ssh_config_file_path, base_driver, test_data):
+def test_resolve_ssh_config(fs_, real_ssh_config_file_path, base_driver, test_data):
     input_data, expected_output, mount_real_file, fake_fs_destination = test_data
 
     if mount_real_file:
-        fs.add_real_file(source_path=real_ssh_config_file_path, target_path=fake_fs_destination)
+        fs_.add_real_file(source_path=real_ssh_config_file_path, target_path=fake_fs_destination)
     actual_output = base_driver._resolve_ssh_config(ssh_config_file=input_data)
     assert actual_output == expected_output
 
@@ -269,11 +267,11 @@ def test_resolve_ssh_config(fs, real_ssh_config_file_path, base_driver, test_dat
     ],
     ids=("auto_etc", "auto_user", "manual_location", "no_config"),
 )
-def test_resolve_ssh_known_hosts(fs, real_ssh_known_hosts_file_path, base_driver, test_data):
+def test_resolve_ssh_known_hosts(fs_, real_ssh_known_hosts_file_path, base_driver, test_data):
     input_data, expected_output, mount_real_file, fake_fs_destination = test_data
 
     if mount_real_file:
-        fs.add_real_file(
+        fs_.add_real_file(
             source_path=real_ssh_known_hosts_file_path, target_path=fake_fs_destination
         )
     actual_output = base_driver._resolve_ssh_known_hosts(ssh_known_hosts=input_data)

tests/unit/driver/generic/test_generic_async_driver.py

@@ -56,9 +56,9 @@ async def _send_input(cls, **kwargs):
 
 
 async def test_send_commands_from_file(
-    fs, monkeypatch, real_ssh_commands_file_path, async_generic_driver
+    fs_, monkeypatch, real_ssh_commands_file_path, async_generic_driver
 ):
-    fs.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
+    fs_.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
 
     async def _send_input(cls, **kwargs):
         return b"raw", b"processed"

tests/unit/driver/generic/test_generic_base_driver.py

@@ -148,8 +148,8 @@ def test_pre_send_commands(base_generic_driver):
     assert isinstance(actual_responses, MultiResponse)
 
 
-def test_pre_send_from_file(fs, real_ssh_config_file_path, base_generic_driver):
-    fs.add_real_file(source_path=real_ssh_config_file_path, target_path="/scrapli/mycommands")
+def test_pre_send_from_file(fs_, real_ssh_config_file_path, base_generic_driver):
+    fs_.add_real_file(source_path=real_ssh_config_file_path, target_path="/scrapli/mycommands")
     commands = base_generic_driver._pre_send_from_file(
         file="/scrapli/mycommands", caller="commands"
     )

tests/unit/driver/generic/test_generic_sync_driver.py

@@ -52,8 +52,10 @@ def test_send_commands(monkeypatch, sync_generic_driver):
     assert actual_response[0].raw_result == b"raw"
 
 
-def test_send_commands_from_file(fs, monkeypatch, real_ssh_commands_file_path, sync_generic_driver):
-    fs.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
+def test_send_commands_from_file(
+    fs_, monkeypatch, real_ssh_commands_file_path, sync_generic_driver
+):
+    fs_.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
     monkeypatch.setattr(
         "scrapli.channel.sync_channel.Channel.send_input",
         lambda _, **kwargs: (b"raw", b"processed"),

tests/unit/driver/network/test_network_async_driver.py

@@ -323,9 +323,9 @@ async def _send_input(cls, channel_input, **kwargs):
 
 
 async def test_send_commands_from_file(
-    fs, monkeypatch, real_ssh_commands_file_path, async_network_driver
+    fs_, monkeypatch, real_ssh_commands_file_path, async_network_driver
 ):
-    fs.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
+    fs_.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
 
     async def _acquire_appropriate_privilege_level(cls, **kwargs):
         return
@@ -449,9 +449,9 @@ async def _send_input(cls, channel_input, **kwargs):
 
 
 async def test_send_configs_from_file(
-    fs, monkeypatch, real_ssh_commands_file_path, async_network_driver
+    fs_, monkeypatch, real_ssh_commands_file_path, async_network_driver
 ):
-    fs.add_real_file(source_path=real_ssh_commands_file_path, target_path="/configs")
+    fs_.add_real_file(source_path=real_ssh_commands_file_path, target_path="/configs")
 
     async def _acquire_priv(cls, **kwargs):
         return

tests/unit/driver/network/test_network_sync_driver.py

@@ -305,8 +305,10 @@ def _send_input(cls, channel_input, **kwargs):
     assert actual_response[0].raw_result == b"raw"
 
 
-def test_send_commands_from_file(fs, monkeypatch, real_ssh_commands_file_path, sync_network_driver):
-    fs.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
+def test_send_commands_from_file(
+    fs_, monkeypatch, real_ssh_commands_file_path, sync_network_driver
+):
+    fs_.add_real_file(source_path=real_ssh_commands_file_path, target_path="/commands")
 
     def _acquire_appropriate_privilege_level(cls, **kwargs):
         return
@@ -419,8 +421,8 @@ def _send_input(cls, channel_input, **kwargs):
     assert actual_response.raw_result == b""
 
 
-def test_send_configs_from_file(fs, monkeypatch, real_ssh_commands_file_path, sync_network_driver):
-    fs.add_real_file(source_path=real_ssh_commands_file_path, target_path="/configs")
+def test_send_configs_from_file(fs_, monkeypatch, real_ssh_commands_file_path, sync_network_driver):
+    fs_.add_real_file(source_path=real_ssh_commands_file_path, target_path="/configs")
 
     def _acquire_priv(cls, **kwargs):
         return

tests/unit/test_helper.py

@@ -142,7 +142,7 @@ def test_textfsm_parse_failed_to_parse():
 
 
 @pytest.mark.skipif(
-    sys.version_info.minor > 9, reason="genie not currently available for python 3.10"
+    sys.version_info.minor > 10, reason="genie not currently available for python 3.11"
 )
 def test_genie_parser():
     result = genie_parse("iosxe", "show ip arp", IOS_ARP)
@@ -153,13 +153,11 @@ def test_genie_parser():
 
 
 @pytest.mark.skipif(
-    sys.version_info.minor > 9, reason="genie not currently available for python 3.10"
+    sys.version_info.minor > 10, reason="genie not currently available for python 3.11"
 )
 def test_genie_parse_failure():
     result = genie_parse("iosxe", "show ip arp", "not really arp data")
     assert result == []
-    # w/out killing this module pyfakefs explodes. dont remember why/how i found that out...
-    del sys.modules["pyats.configuration"]
 
 
 def test_genie_no_genie_installed(monkeypatch):
@@ -240,17 +238,15 @@ def _import_module(name):
     assert output == []
 
 
-def test_resolve_file(fs, real_ssh_config_file_path):
+def test_resolve_file(fs_, real_ssh_config_file_path):
     # pyfakefs so this is not host dependent
-    _ = fs
-    fs.add_real_file(source_path=real_ssh_config_file_path, target_path="/some/neat/path/myfile")
+    fs_.add_real_file(source_path=real_ssh_config_file_path, target_path="/some/neat/path/myfile")
     assert resolve_file(file="/some/neat/path/myfile") == "/some/neat/path/myfile"
 
 
-def test_resolve_file_expanduser(fs, real_ssh_config_file_path):
+def test_resolve_file_expanduser(fs_, real_ssh_config_file_path):
     # pyfakefs so this is not host dependent
-    _ = fs
-    fs.add_real_file(
+    fs_.add_real_file(
         source_path=real_ssh_config_file_path, target_path=Path("~/myfile").expanduser()
     )
     assert resolve_file(file="~/myfile") == str(Path("~/myfile").expanduser())

tests/unit/test_logging.py

@@ -61,7 +61,7 @@ def test_scrapli_filehandler():
     pass
 
 
-def test_enable_basic_logging(fs):
+def test_enable_basic_logging(fs_):
     assert Path("scrapli.log").is_file() is False
     enable_basic_logging(file=True, level="debug")
     scrapli_logger = logging.getLogger("scrapli")
@@ -78,7 +78,7 @@ def test_enable_basic_logging(fs):
     del logger.handlers[1]
 
 
-def test_enable_basic_logging_no_buffer(fs):
+def test_enable_basic_logging_no_buffer(fs_):
     assert Path("mylog.log").is_file() is False
 
     enable_basic_logging(file="mylog.log", level="debug", buffer_log=False, caller_info=True)

tests/unit/test_response.py

@@ -163,7 +163,7 @@ def test_response_parse_textfsm_no_template():
 
 
 @pytest.mark.skipif(
-    sys.version_info.minor > 9, reason="genie not currently available for python 3.10"
+    sys.version_info.minor > 10, reason="genie not currently available for python 3.11"
 )
 def test_response_parse_genie():
     response = Response("localhost", channel_input="show ip arp", genie_platform="iosxe")
@@ -179,7 +179,7 @@ def test_response_parse_genie():
 
 
 @pytest.mark.skipif(
-    sys.version_info.minor > 9, reason="genie not currently available for python 3.10"
+    sys.version_info.minor > 10, reason="genie not currently available for python 3.11"
 )
 def test_response_parse_genie_fail():
     response = Response("localhost", channel_input="show ip arp", genie_platform="iosxe")

tests/unit/test_ssh_config.py

@@ -57,7 +57,7 @@ def test_init_ssh_config_file_explicit(real_ssh_config_file_path):
     assert ssh_conf.ssh_config == ssh_config_file
 
 
-def test_init_ssh_config_file_no_config_file(fs):
+def test_init_ssh_config_file_no_config_file(fs_):
     ssh_conf = SSHConfig("")
     # should only have a single splat host w/ all values set to None/empty
     assert ["*"] == list(ssh_conf.hosts.keys())
@@ -164,7 +164,7 @@ def test_init_ssh_known_hosts_file_explicit(real_ssh_known_hosts_file_path):
     assert known_hosts.ssh_known_hosts == ssh_known_hosts
 
 
-def test_init_ssh_known_hosts_file_no_config_file(fs):
+def test_init_ssh_known_hosts_file_no_config_file(fs_):
     known_hosts = SSHKnownHosts("")
     assert known_hosts.hosts == {}
 

tests/unit/transport/plugins/system/test_system_transport.py

@@ -84,7 +84,7 @@ def test_build_open_cmd_alternate_options(system_transport):
     ]
 
 
-def test_close(fs, monkeypatch, system_transport):
+def test_close(fs_, monkeypatch, system_transport):
     def _close(cls):
         pass
 
@@ -95,7 +95,7 @@ def _close(cls):
 
     # giving ptyprocess a "real" (but not like... real real) fd seemed like a good idea... dunno
     # if its really necessary, but it *does* need a fd of some sort so whatever
-    fs.create_file("dummy")
+    fs_.create_file("dummy")
     dummy_file = open("dummy")
     system_transport.session = PtyProcess(pid=0, fd=dummy_file.fileno())
     system_transport.close()
@@ -107,21 +107,21 @@ def test_isalive_no_session(system_transport):
     assert system_transport.isalive() is False
 
 
-def test_isalive(fs, system_transport):
+def test_isalive(fs_, system_transport):
     # lie and pretend the session is already assigned
     # giving ptyprocess a "real" (but not like... real real) fd seemed like a good idea... dunno
     # if its really necessary, but it *does* need a fd of some sort so whatever; also give it a
     # forked pid so that the isalive method works... obviously this is sorta cheating to force it
     # to work but we really only care that scrapli does the right thing... we have faith that
     # ptyprocess will be doing the right thing "below" scrapli
     dummy_pid, fd = pty.fork()
-    fs.create_file("dummy")
+    fs_.create_file("dummy")
     dummy_file = open("dummy")
     system_transport.session = PtyProcess(pid=dummy_pid, fd=dummy_file.fileno())
     assert system_transport.isalive() is True
 
 
-def test_read(fs, monkeypatch, system_transport):
+def test_read(fs_, monkeypatch, system_transport):
     def _read(cls, _):
         return b"somebytes"
 
@@ -134,7 +134,7 @@ def _read(cls, _):
     # giving ptyprocess a "real" (but not like... real real) fd seemed like a good idea... dunno
     # if its really necessary, but it *does* need a fd of some sort so whatever
     dummy_pid, fd = pty.fork()
-    fs.create_file("dummy")
+    fs_.create_file("dummy")
     dummy_file = open("dummy")
     system_transport.session = PtyProcess(pid=dummy_pid, fd=dummy_file.fileno())
 
@@ -146,7 +146,7 @@ def test_read_exception_not_open(system_transport):
         system_transport.read()
 
 
-def test_read_exception_eof(fs, monkeypatch, system_transport):
+def test_read_exception_eof(fs_, monkeypatch, system_transport):
     def _read(cls, _):
         raise EOFError
 
@@ -158,7 +158,7 @@ def _read(cls, _):
     # lie and pretend the session is already assigned
     # giving ptyprocess a "real" (but not like... real real) fd seemed like a good idea... dunno
     # if its really necessary, but it *does* need a fd of some sort so whatever
-    fs.create_file("dummy")
+    fs_.create_file("dummy")
     dummy_file = open("dummy")
     system_transport.session = PtyProcess(pid=0, fd=dummy_file.fileno())