quickjs-tart

quickjs-based runtime for wallet-core logic
Log | Files | Refs | README | LICENSE

test_11_unix.py (4472B)

      1 #!/usr/bin/env python3
      2 # -*- coding: utf-8 -*-
      3 #***************************************************************************
      4 #                                  _   _ ____  _
      5 #  Project                     ___| | | |  _ \| |
      6 #                             / __| | | | |_) | |
      7 #                            | (__| |_| |  _ <| |___
      8 #                             \___|\___/|_| \_\_____|
      9 #
     10 # Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.
     11 #
     12 # This software is licensed as described in the file COPYING, which
     13 # you should have received as part of this distribution. The terms
     14 # are also available at https://curl.se/docs/copyright.html.
     15 #
     16 # You may opt to use, copy, modify, merge, publish, distribute and/or sell
     17 # copies of the Software, and permit persons to whom the Software is
     18 # furnished to do so, under the terms of the COPYING file.
     19 #
     20 # This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
     21 # KIND, either express or implied.
     22 #
     23 # SPDX-License-Identifier: curl
     24 #
     25 ###########################################################################
     26 #
     27 import logging
     28 import os
     29 import socket
     30 from threading import Thread
     31 from typing import Generator
     32 
     33 import pytest
     34 
     35 from testenv import Env, CurlClient
     36 
     37 
     38 log = logging.getLogger(__name__)
     39 
     40 
     41 class UDSFaker:
     42 
     43     def __init__(self, path):
     44         self._uds_path = path
     45         self._done = False
     46         self._socket = None
     47         self._thread = None
     48 
     49     @property
     50     def path(self):
     51         return self._uds_path
     52 
     53     def start(self):
     54         def process(self):
     55             self._socket.listen(1)
     56             self._process()
     57 
     58         try:
     59             os.unlink(self._uds_path)
     60         except OSError:
     61             if os.path.exists(self._uds_path):
     62                 raise
     63         self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     64         self._socket.bind(self._uds_path)
     65         self._thread = Thread(target=process, daemon=True, args=[self])
     66         self._thread.start()
     67 
     68     def stop(self):
     69         self._done = True
     70         self._socket.close()
     71 
     72     def _process(self):
     73         while self._done is False:
     74             try:
     75                 c, client_address = self._socket.accept()
     76                 try:
     77                     c.recv(16)
     78                     c.sendall("""HTTP/1.1 200 Ok
     79 Server: UdsFaker
     80 Content-Type: application/json
     81 Content-Length: 19
     82 
     83 { "host": "faked" }""".encode())
     84                 finally:
     85                     c.close()
     86 
     87             except ConnectionAbortedError:
     88                 self._done = True
     89             except OSError:
     90                 self._done = True
     91 
     92 
     93 class TestUnix:
     94 
     95     @pytest.fixture(scope="class")
     96     def uds_faker(self, env: Env) -> Generator[UDSFaker, None, None]:
     97         uds_path = os.path.join(env.gen_dir, 'uds_11.sock')
     98         faker = UDSFaker(path=uds_path)
     99         faker.start()
    100         yield faker
    101         faker.stop()
    102 
    103     # download http: via Unix socket
    104     def test_11_01_unix_connect_http(self, env: Env, httpd, uds_faker):
    105         curl = CurlClient(env=env)
    106         url = f'http://{env.domain1}:{env.http_port}/data.json'
    107         r = curl.http_download(urls=[url], with_stats=True,
    108                                extra_args=[
    109                                  '--unix-socket', uds_faker.path,
    110                                ])
    111         r.check_response(count=1, http_status=200)
    112 
    113     # download https: via Unix socket
    114     @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    115     def test_11_02_unix_connect_http(self, env: Env, httpd, uds_faker):
    116         curl = CurlClient(env=env)
    117         url = f'https://{env.domain1}:{env.https_port}/data.json'
    118         r = curl.http_download(urls=[url], with_stats=True,
    119                                extra_args=[
    120                                  '--unix-socket', uds_faker.path,
    121                                ])
    122         r.check_response(exitcode=35, http_status=None)
    123 
    124     # download HTTP/3 via Unix socket
    125     @pytest.mark.skipif(condition=not Env.have_h3(), reason='h3 not supported')
    126     def test_11_03_unix_connect_quic(self, env: Env, httpd, uds_faker):
    127         curl = CurlClient(env=env)
    128         url = f'https://{env.domain1}:{env.https_port}/data.json'
    129         r = curl.http_download(urls=[url], with_stats=True,
    130                                alpn_proto='h3',
    131                                extra_args=[
    132                                  '--unix-socket', uds_faker.path,
    133                                ])
    134         r.check_response(exitcode=96, http_status=None)