Commit fixes recommended by pyupgrade tool

asyncssh/agent.py

@@ -252,7 +252,7 @@ async def _make_request(self, msgtype: int, *args: bytes) -> \
 
                 resplen = int.from_bytes((await reader.readexactly(4)), 'big')
 
-                resp = SSHPacket((await reader.readexactly(resplen)))
+                resp = SSHPacket(await reader.readexactly(resplen))
                 resptype = resp.get_byte()
 
                 return resptype, resp

asyncssh/auth.py

@@ -181,7 +181,7 @@ async def _start(self) -> None:
 
             self._gss = self._conn.get_gss_context()
             self._gss.reset()
-            mechs = b''.join((String(mech) for mech in self._gss.mechs))
+            mechs = b''.join(String(mech) for mech in self._gss.mechs)
             await self.send_request(UInt32(len(self._gss.mechs)), mechs)
         else:
             self._conn.try_next_auth()

asyncssh/connection.py

@@ -725,7 +725,7 @@ def get_port(self) -> int:
         if hasattr(self._server, 'get_port'):
             return self._server.get_port()
         else:
-            ports = set(addr[1] for addr in self.get_addresses())
+            ports = {addr[1] for addr in self.get_addresses()}
             return ports.pop() if len(ports) == 1 else 0
 
     def close(self) -> None:

asyncssh/crypto/x509.py

@@ -94,8 +94,8 @@ def _to_purpose_oids(purposes: _Purposes) -> _PurposeOIDs:
     if not purposes or 'any' in purposes or _purpose_any in purposes:
         purpose_oids = None
     else:
-        purpose_oids = set(_purpose_to_oid.get(p) or x509.ObjectIdentifier(p)
-                           for p in purposes)
+        purpose_oids = {_purpose_to_oid.get(p) or x509.ObjectIdentifier(p)
+                           for p in purposes}
 
     return purpose_oids
 
@@ -143,8 +143,8 @@ class X509Name(x509.Name):
         ('CN', x509.NameOID.COMMON_NAME),
         ('DC', x509.NameOID.DOMAIN_COMPONENT))
 
-    _to_oid = dict((k, v) for k, v in _attrs)
-    _from_oid = dict((v, k) for k, v in _attrs)
+    _to_oid = dict(_attrs)
+    _from_oid = {v: k for k, v in _attrs}
 
     def __init__(self, name: _NameInit):
         if isinstance(name, str):

asyncssh/known_hosts.py

@@ -208,8 +208,8 @@ def _match(self, host: str, addr: str,
                 ip = None
 
         if port:
-            host = '[{}]:{}'.format(host, port) if host else ''
-            addr = '[{}]:{}'.format(addr, port) if addr else ''
+            host = f'[{host}]:{port}' if host else ''
+            addr = f'[{addr}]:{port}' if addr else ''
 
         matches = []
         matches += self._exact_entries.get(host, [])

asyncssh/public_key.py

@@ -1941,8 +1941,8 @@ def validate_chain(self, trust_chain: Sequence['SSHX509Certificate'],
                        host_principal: str = '') -> None:
         """Validate an X.509 certificate chain"""
 
-        trust_store = set(c for c in trust_chain if c.subject != c.issuer) | \
-            set(c for c in trusted_certs)
+        trust_store = {c for c in trust_chain if c.subject != c.issuer} | \
+            set(trusted_certs)
 
         if trusted_cert_paths:
             self._expand_trust_store(self, trusted_cert_paths, trust_store)

asyncssh/sftp.py

@@ -1628,7 +1628,7 @@ def _format(self, k: str, v: object) -> Optional[str]:
             return _file_types.get(cast(int, v), str(v)) \
                 if v != FILEXFER_TYPE_UNKNOWN else None
         elif k == 'permissions':
-            return '{:04o}'.format(cast(int, v))
+            return f'{cast(int, v):04o}'
         elif k in ('atime', 'crtime', 'mtime', 'ctime'):
             return self._format_ns(k)
         elif k in ('atime_ns', 'crtime_ns', 'mtime_ns', 'ctime_ns'):

examples/simple_keyed_server.py

@@ -42,7 +42,7 @@ def connection_made(self, conn: asyncssh.SSHServerConnection) -> None:
     def begin_auth(self, username: str) -> bool:
         try:
             self._conn.set_authorized_keys('authorized_keys/%s' % username)
-        except IOError:
+        except OSError:
             pass
 
         return True

tests/test_channel.py

@@ -271,7 +271,7 @@ async def _begin_session(self, stdin, stdout, stderr):
         elif action == 'agent':
             try:
                 async with asyncssh.connect_agent(self._conn) as agent:
-                    stdout.write(str(len((await agent.get_keys()))) + '\n')
+                    stdout.write(str(len(await agent.get_keys())) + '\n')
             except (OSError, asyncssh.ChannelOpenError):
                 stdout.channel.exit(1)
         elif action == 'agent_sock':
@@ -280,7 +280,7 @@ async def _begin_session(self, stdin, stdout, stderr):
             if agent_path:
                 async with asyncssh.connect_agent(agent_path) as agent:
                     await asyncio.sleep(0.1)
-                    stdout.write(str(len((await agent.get_keys()))) + '\n')
+                    stdout.write(str(len(await agent.get_keys())) + '\n')
             else:
                 stdout.channel.exit(1)
         elif action == 'rejected_agent':
@@ -428,11 +428,11 @@ async def _begin_session(self, stdin, stdout, stderr):
         elif action == 'empty_data':
             stdin.channel.send_packet(MSG_CHANNEL_DATA, String(''))
         elif action == 'partial_unicode':
-            data = '\xff\xff'.encode('utf-8')
+            data = '\xff\xff'.encode()
             stdin.channel.send_packet(MSG_CHANNEL_DATA, String(data[:3]))
             stdin.channel.send_packet(MSG_CHANNEL_DATA, String(data[3:]))
         elif action == 'partial_unicode_at_eof':
-            data = '\xff\xff'.encode('utf-8')
+            data = '\xff\xff'.encode()
             stdin.channel.send_packet(MSG_CHANNEL_DATA, String(data[:3]))
         elif action == 'unicode_error':
             stdin.channel.send_packet(MSG_CHANNEL_DATA, String(b'\xff'))
@@ -767,7 +767,7 @@ async def test_unknown_channel_request(self):
             async with self.connect() as conn:
                 chan, _ = await _create_session(conn)
 
-                self.assertFalse((await chan.make_request('unknown')))
+                self.assertFalse(await chan.make_request('unknown'))
 
     @asynctest
     async def test_invalid_channel_request(self):

tests/test_connection.py

@@ -798,7 +798,7 @@ async def test_import_known_hosts(self):
 
         known_hosts_path = os.path.join('.ssh', 'known_hosts')
 
-        with open(known_hosts_path, 'r') as f:
+        with open(known_hosts_path) as f:
             known_hosts = asyncssh.import_known_hosts(f.read())
 
         async with self.connect(known_hosts=known_hosts):

tests/test_process.py

@@ -460,7 +460,7 @@ async def test_ignoring_invalid_unicode(self):
     async def test_incomplete_unicode(self):
         """Test incomplete Unicode data"""
 
-        data = '\u2000'.encode('utf-8')[:2]
+        data = '\u2000'.encode()[:2]
 
         with open('stdin', 'wb') as file:
             file.write(data)
@@ -586,7 +586,7 @@ async def test_stdin_open_file(self):
         with open('stdin', 'w') as file:
             file.write(data)
 
-        file = open('stdin', 'r')
+        file = open('stdin')
 
         async with self.connect() as conn:
             result = await conn.run('echo', stdin=file)
@@ -773,7 +773,7 @@ async def test_stdout_file(self):
         async with self.connect() as conn:
             result = await conn.run('echo', input=data, stdout='stdout')
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             stdout_data = file.read()
 
         self.assertEqual(stdout_data, data)
@@ -806,7 +806,7 @@ async def test_stdout_pathlib(self):
         async with self.connect() as conn:
             result = await conn.run('echo', input=data, stdout=Path('stdout'))
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             stdout_data = file.read()
 
         self.assertEqual(stdout_data, data)
@@ -824,7 +824,7 @@ async def test_stdout_open_file(self):
         async with self.connect() as conn:
             result = await conn.run('echo', input=data, stdout=file)
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             stdout_data = file.read()
 
         self.assertEqual(stdout_data, data)
@@ -842,7 +842,7 @@ async def test_stdout_open_file_keep_open(self):
                 await conn.run('echo', input=data, stdout=file, recv_eof=False)
                 await conn.run('echo', input=data, stdout=file, recv_eof=False)
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             stdout_data = file.read()
 
         self.assertEqual(stdout_data, 2*data)
@@ -995,7 +995,7 @@ async def test_change_stdout(self):
 
             result = await process.wait()
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             stdout_data = file.read()
 
         self.assertEqual(stdout_data, 'xxx')
@@ -1246,7 +1246,7 @@ async def test_stdout_aiofile(self):
         async with self.connect() as conn:
             result = await conn.run('echo', input=data, stdout=file)
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             stdout_data = file.read()
 
         self.assertEqual(stdout_data, data)
@@ -1264,7 +1264,7 @@ async def test_stdout_aiofile_keep_open(self):
                 await conn.run('echo', input=data, stdout=file, recv_eof=False)
                 await conn.run('echo', input=data, stdout=file, recv_eof=False)
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             stdout_data = file.read()
 
         self.assertEqual(stdout_data, 2*data)
@@ -1316,7 +1316,7 @@ async def test_pause_async_file_writer(self):
                 await conn.run('delay', input=data, stdout=file,
                                stderr=asyncssh.DEVNULL)
 
-        with open('stdout', 'r') as file:
+        with open('stdout') as file:
             self.assertEqual(file.read(), data)
 
 

tests/test_public_key.py

@@ -180,7 +180,7 @@ def select_passphrase(cipher, pbe_version=0):
                                          'rc4-40', 'rc4-128'):
         return 'passphrase'.encode('utf-16-be')
     else:
-        return 'passphrase'.encode('utf-8')
+        return b'passphrase'
 
 
 class _TestPublicKey(TempDirTestCase):