(
self, server_port: int, client_certificate: PrivateCertificate
)
| 238 | |
| 239 | @async_yield_fixture # type: ignore[untyped-decorator] |
| 240 | async def client( |
| 241 | self, server_port: int, client_certificate: PrivateCertificate |
| 242 | ) -> AsyncGenerator[H2ClientProtocol]: |
| 243 | from twisted.internet import reactor |
| 244 | |
| 245 | from scrapy.core.http2.protocol import H2ClientFactory # noqa: PLC0415 |
| 246 | |
| 247 | client_options = optionsForClientTLS( |
| 248 | hostname=self.host, |
| 249 | trustRoot=client_certificate, |
| 250 | acceptableProtocols=[b"h2"], |
| 251 | ) |
| 252 | uri = URI.fromBytes(bytes(self.get_url(server_port, "/"), "utf-8")) |
| 253 | h2_client_factory = H2ClientFactory(uri, Settings(), Deferred()) |
| 254 | client_endpoint = SSL4ClientEndpoint( |
| 255 | reactor, self.host, server_port, client_options |
| 256 | ) |
| 257 | client = await client_endpoint.connect(h2_client_factory) |
| 258 | |
| 259 | yield client |
| 260 | |
| 261 | if client.connected: |
| 262 | client.transport.loseConnection() |
| 263 | client.transport.abortConnection() |
| 264 | |
| 265 | def get_url(self, portno: int, path: str) -> str: |
| 266 | """ |
no test coverage detected