mirror of
https://github.com/tahoe-lafs/tahoe-lafs.git
synced 2025-02-21 10:01:54 +00:00
More type annotations.
This commit is contained in:
parent
0d0e32646f
commit
00b7e7e178
@ -17,11 +17,13 @@ from eliot import start_action
|
||||
from cryptography.x509 import Certificate as CryptoCertificate
|
||||
from zope.interface import implementer
|
||||
from klein import Klein, KleinRenderable
|
||||
from klein.resource import KleinResource
|
||||
from twisted.web import http
|
||||
from twisted.internet.interfaces import (
|
||||
IListeningPort,
|
||||
IStreamServerEndpoint,
|
||||
IPullProducer,
|
||||
IProtocolFactory,
|
||||
)
|
||||
from twisted.internet.address import IPv4Address, IPv6Address
|
||||
from twisted.internet.defer import Deferred
|
||||
@ -668,7 +670,7 @@ class HTTPServer(BaseApp):
|
||||
self._uploads.remove_write_bucket
|
||||
)
|
||||
|
||||
def get_resource(self):
|
||||
def get_resource(self) -> KleinResource:
|
||||
"""Return twisted.web ``Resource`` for this object."""
|
||||
return self._app.resource()
|
||||
|
||||
@ -1085,7 +1087,10 @@ class _TLSEndpointWrapper(object):
|
||||
|
||||
@classmethod
|
||||
def from_paths(
|
||||
cls, endpoint, private_key_path: FilePath, cert_path: FilePath
|
||||
cls: type[_TLSEndpointWrapper],
|
||||
endpoint: IStreamServerEndpoint,
|
||||
private_key_path: FilePath,
|
||||
cert_path: FilePath,
|
||||
) -> "_TLSEndpointWrapper":
|
||||
"""
|
||||
Create an endpoint with the given private key and certificate paths on
|
||||
@ -1100,7 +1105,7 @@ class _TLSEndpointWrapper(object):
|
||||
)
|
||||
return cls(endpoint=endpoint, context_factory=certificate_options)
|
||||
|
||||
def listen(self, factory):
|
||||
def listen(self, factory: IProtocolFactory) -> Deferred[IListeningPort]:
|
||||
return self.endpoint.listen(
|
||||
TLSMemoryBIOFactory(self.context_factory, False, factory)
|
||||
)
|
||||
|
@ -109,9 +109,11 @@ class PinningHTTPSValidation(AsyncTestCase):
|
||||
root.isLeaf = True
|
||||
listening_port = await endpoint.listen(Site(root))
|
||||
try:
|
||||
yield f"https://127.0.0.1:{listening_port.getHost().port}/"
|
||||
yield f"https://127.0.0.1:{listening_port.getHost().port}/" # type: ignore[attr-defined]
|
||||
finally:
|
||||
await listening_port.stopListening()
|
||||
result = listening_port.stopListening()
|
||||
if result is not None:
|
||||
await result
|
||||
|
||||
def request(self, url: str, expected_certificate: x509.Certificate):
|
||||
"""
|
||||
|
Loading…
x
Reference in New Issue
Block a user