Best Python code snippet using playwright-python
https_server.py
Source:https_server.py  
1"""2//  -------------------------------------------------------------3//  author        Giga4//  project       qeeqbox/honeypots5//  email         gigaqeeq@gmail.com6//  description   app.py (CLI)7//  licensee      AGPL-3.08//  -------------------------------------------------------------9//  contributors list qeeqbox/social-analyzer/graphs/contributors10//  -------------------------------------------------------------11"""12from uuid import uuid413from honeypots.helper import close_port_wrapper, get_free_port, kill_server_wrapper, server_arguments, setup_logger, disable_logger14from os import path15from subprocess import Popen16from twisted.python import log as tlog17from random import choice18from twisted.web.resource import Resource19from twisted.web.server import Site20from twisted.internet import reactor, ssl21from tempfile import gettempdir, _get_candidate_names22from time import sleep23from requests.packages.urllib3 import disable_warnings24from requests import get, post25from cgi import FieldStorage26from OpenSSL import crypto27from warnings import filterwarnings28filterwarnings(action='ignore', module='.*OpenSSL.*')29disable_warnings()30class QHTTPSServer():31    def __init__(self, ip=None, port=None, username=None, password=None, mocking=False, config=''):32        self.ip = ip or '0.0.0.0'33        self.port = port or 44334        self.username = username or "test"35        self.password = password or "test"36        self.mocking = mocking or ''37        self.key = path.join(gettempdir(), next(_get_candidate_names()))38        self.cert = path.join(gettempdir(), next(_get_candidate_names()))39        self.random_servers = ['Apache', 'nginx', 'Microsoft-IIS/7.5', 'Microsoft-HTTPAPI/2.0', 'Apache/2.2.15', 'SmartXFilter', 'Microsoft-IIS/8.5', 'Apache/2.4.6', 'Apache-Coyote/1.1',40                               'Microsoft-IIS/7.0', 'Apache/2.4.18', 'AkamaiGHost', 'Apache/2.2.25', 'Microsoft-IIS/10.0', 'Apache/2.2.3', 'nginx/1.12.1', 'Apache/2.4.29', 'cloudflare', 'Apache/2.2.22']41        self.process = None42        self.uuid = 'honeypotslogger'43        self.config = config44        if config:45            self.logs = setup_logger(self.uuid, config)46        else:47            self.logs = setup_logger(self.uuid, None)48        disable_logger(1, tlog)49    def CreateCert(self, host_name, key, cert):50        pk = crypto.PKey()51        pk.generate_key(crypto.TYPE_RSA, 2048)52        c = crypto.X509()53        c.get_subject().C = 'US'54        c.get_subject().ST = 'OR'55        c.get_subject().L = 'None'56        c.get_subject().O = 'None'57        c.get_subject().OU = 'None'58        c.get_subject().CN = next(_get_candidate_names())59        c.set_serial_number(0)60        before, after = (0, 60 * 60 * 24 * 365 * 2)61        c.gmtime_adj_notBefore(before)62        c.gmtime_adj_notAfter(after)63        c.set_issuer(c.get_subject())64        c.set_pubkey(pk)65        c.sign(pk, 'sha256')66        open(cert, "wb").write(crypto.dump_certificate(crypto.FILETYPE_PEM, c))67        open(key, "wb").write(crypto.dump_privatekey(crypto.FILETYPE_PEM, pk))68    def https_server_main(self):69        _q_s = self70        class MainResource(Resource):71            isLeaf = True72            home_file = b'''73<!DOCTYPE html>74<html>75	<head>76		<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.0.0-beta.3/css/bootstrap.min.css" />77		<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/font-awesome/4.7.0/css/font-awesome.min.css" />78		<meta http-equiv="content-type" content="text/html;charset=utf-8" />79		<title>Login</title>80		<style>81			body,html{height: 100%;text-align: center;},82		</style>83	</head>84	<body>85		<div class="container-fluid h-100">86			<div class="row justify-content-center h-100 align-items-center">87			<div class="col col-xl-3">88				<b>We'll back soon..</b>89			</div>90			</div>91		</div>92	</body>93</html>'''94            login_file = b'''<!DOCTYPE html>95<html>96	<head>97		<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.0.0-beta.3/css/bootstrap.min.css" />98		<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/font-awesome/4.7.0/css/font-awesome.min.css" />99		<meta http-equiv="content-type" content="text/html;charset=utf-8" />100		<title>Login</title>101		<style>body,html {height: 100%;}</style>102	</head>103	<body>104		<div class="container-fluid h-100">105			<div class="row justify-content-center h-100 align-items-center">106			<div class="col col-xl-3">107				<form id="login" action="" method="post">108					<div class="form-group">109						<input class="form-control form-control-sm" name="username" type="text" placeholder="username" id="username">110					</div>111					<div class="form-group">112						<input class="form-control form-control-sm" name="password" type="password" placeholder="password" id="password">113					</div>114					<div class="form-group">115						<button class="btn btn-default btn-sm btn-block" type="submit">login</button>116					</div>117				</form>118			</div>119			</div>120		</div>121	</body>122</html>123'''124            if isinstance(_q_s.mocking, bool):125                if _q_s.mocking == True:126                    server = choice(_q_s.random_servers)127            elif isinstance(_q_s.mocking, str):128                server = _q_s.mocking129            def render(self, request):130                headers = {}131                try:132                    def check_bytes(string):133                        if isinstance(string, bytes):134                            return string.decode()135                        else:136                            return str(string)137                    for item, value in dict(request.requestHeaders.getAllRawHeaders()).items():138                        headers.update(139                            {check_bytes(item): ','.join(map(check_bytes, value))})140                except BaseException:141                    pass142                _q_s.logs.info({'server': 'https_server', 'action': 'connection',143                                'ip': request.getClientIP(), 'request': headers})144                if self.server != "":145                    request.responseHeaders.removeHeader("Server")146                    request.responseHeaders.addRawHeader("Server", self.server)147                if request.method == b"GET":148                    _q_s.logs.info(149                        {'server': 'https_server', 'action': 'get', 'ip': request.getClientIP()})150                    if request.uri == b"/login.html":151                        if _q_s.username != '' and _q_s.password != '':152                            request.responseHeaders.addRawHeader(153                                "Content-Type", "text/html; charset=utf-8")154                            return self.login_file155                    request.responseHeaders.addRawHeader(156                        "Content-Type", "text/html; charset=utf-8")157                    return self.home_file158                elif request.method == b"POST":159                    self.headers = request.getAllHeaders()160                    _q_s.logs.info(161                        {'server': 'https_server', 'action': 'post', 'ip': request.getClientIP()})162                    if request.uri == b"/login.html" or b'/':163                        if _q_s.username != '' and _q_s.password != '':164                            form = FieldStorage(fp=request.content, headers=self.headers, environ={165                                                'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': self.headers[b'content-type'], })166                            if 'username' in form and 'password' in form:167                                if form['username'].value == _q_s.username and form['password'].value == _q_s.password:168                                    _q_s.logs.info({'server': 'https_server', 'action': 'login', 'status': 'success', 'ip': request.getClientIP(169                                    ), 'username': _q_s.username, 'password': _q_s.password})170                                else:171                                    _q_s.logs.info({'server': 'https_server', 'action': 'login', 'status': 'failed', 'ip': request.getClientIP(172                                    ), 'username': form['username'].value, 'password': form['password'].value})173                    request.responseHeaders.addRawHeader(174                        "Content-Type", "text/html; charset=utf-8")175                    return self.home_file176                else:177                    request.responseHeaders.addRawHeader(178                        "Content-Type", "text/html; charset=utf-8")179                    return self.home_file180        self.CreateCert("localhost", self.key, self.cert)181        ssl_context = ssl.DefaultOpenSSLContextFactory(self.key, self.cert)182        reactor.listenSSL(self.port, Site(MainResource()), ssl_context)183        reactor.run()184    def run_server(self, process=False, auto=False):185        if process:186            if auto:187                port = get_free_port()188                if port > 0:189                    self.port = port190                    self.process = Popen(['python3', path.realpath(__file__), '--custom', '--ip', str(self.ip), '--port', str(self.port), '--username', str(191                        self.username), '--password', str(self.password), '--mocking', str(self.mocking), '--config', str(self.config), '--uuid', str(self.uuid)])192                    if self.process.poll() is None:193                        self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'success', 'route': '/login.html',194                                        'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})195                    else:196                        self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'error',197                                        'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})198                else:199                    self.logs.info({'server': 'https_server', 'action': 'setup', 'status': 'error',200                                    'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})201            elif self.close_port() and self.kill_server():202                self.process = Popen(['python3', path.realpath(__file__), '--custom', '--ip', str(self.ip), '--port', str(self.port), '--username', str(203                    self.username), '--password', str(self.password), '--mocking', str(self.mocking), '--config', str(self.config), '--uuid', str(self.uuid)])204                if self.process.poll() is None:205                    self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'success', 'route': '/login.html',206                                    'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})207                else:208                    self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'error',209                                    'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})210        else:211            self.https_server_main()212    def test_server(self, ip=None, port=None, username=None, password=None):213        try:214            sleep(2)215            _ip = ip or self.ip216            _port = port or self.port217            _username = username or self.username218            _password = password or self.password219            get('https://{}:{}'.format(_ip, _port), verify=False)220            post('https://{}:{}'.format(_ip, _port), data={'username': (None,221                                                                        _username), 'password': (None, _password)}, verify=False)222        except BaseException:223            pass224    def close_port(self):225        ret = close_port_wrapper('https_server', self.ip, self.port, self.logs)226        return ret227    def kill_server(self):228        ret = kill_server_wrapper('https_server', self.uuid, self.process)229        return ret230if __name__ == '__main__':231    parsed = server_arguments()232    if parsed.docker or parsed.aws or parsed.custom:233        qhttpsserver = QHTTPSServer(ip=parsed.ip, port=parsed.port, username=parsed.username,234                                    password=parsed.password, mocking=parsed.mocking, config=parsed.config)...test_handler.py
Source:test_handler.py  
1import pytest2import base643import json4import requests5import ssl6import threading7from dummyserver import handler8from dummyserver import server9SERVER_CRT = "test/test_handler/server-bundle.crt"10SERVER_CA  = "test/test_handler/rootCA.crt"11CLIENT_CRT = "test/test_handler/client.crt"12CLIENT_KEY = "test/test_handler/client.key"13CLIENT_CA  = "test/test_handler/rootCA.crt"14HTTP_PORT   = 808015HTTPS_PORT  = 818116HTTPSM_PORT = 828217MGMT_PORT   = 838318USERNAME = "testuser"19PASSWORD = "testpass"20CREDENTIALS = base64.b64encode(b"testuser:testpass")21ERROR_RANGE = list(range(402, 418))+list(range(500, 505))22@pytest.fixture(scope="module")23def mgmt_server():24    # Startup25    print("Starting server")26    http_server = server.ThreadedHTTPServer(("", MGMT_PORT), handler.DummyHandler)27    http_thread = threading.Thread(target=http_server.serve_forever)28    http_thread.setDaemon(True)29    http_thread.start()30    yield http_server31    # Teardown32    print("Stopping server")33    http_server.shutdown()34    http_server.server_close()35@pytest.fixture(scope="module")36def http_server_with_credentials():37    # Startup38    print("Starting server")39    http_server = server.ThreadedHTTPServer(("", HTTP_PORT), handler.HTTPHandler)40    http_thread = threading.Thread(target=http_server.serve_forever)41    http_thread.setDaemon(True)42    http_thread.start()43    yield http_server44    # Teardown45    print("Stopping server")46    http_server.shutdown()47    http_server.server_close()48@pytest.fixture(scope="module")49def https_server_with_credentials():50    # Startup51    print("Starting server")52    context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)53    context.verify_mode = ssl.CERT_NONE54    context.load_cert_chain(certfile=SERVER_CRT)55    handler.HTTPHandler.credentials = CREDENTIALS56    https_server = server.ThreadedHTTPServer(("", HTTPS_PORT), handler.HTTPHandler)57    https_server.socket = context.wrap_socket (https_server.socket, server_side=True)58    https_thread = threading.Thread(target=https_server.serve_forever)59    https_thread.setDaemon(True)60    https_thread.start()61    yield https_server62    # Teardown63    print("Stopping server")64    https_server.shutdown()65    https_server.server_close()66@pytest.fixture(scope="module")67def https_server_with_mutual_auth_with_credentials():68    # Startup69    print("Starting server")70    context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)71    context.verify_mode = ssl.CERT_REQUIRED72    context.load_cert_chain(certfile=SERVER_CRT)73    context.load_verify_locations(cafile=CLIENT_CA)74    handler.HTTPHandler.credentials = CREDENTIALS75    https_server = server.ThreadedHTTPServer(("", HTTPSM_PORT), handler.HTTPHandler)76    https_server.socket = context.wrap_socket (https_server.socket, server_side=True)77    https_thread = threading.Thread(target=https_server.serve_forever)78    https_thread.setDaemon(True)79    https_thread.start()80    yield https_server81    # Teardown82    print("Stopping server")83    https_server.shutdown()84    https_server.server_close()85def test_http_with_credentials(http_server_with_credentials):86    r = requests.get(87        url = 'http://localhost:{}'.format(HTTP_PORT),88        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)89    )90    assert r.status_code == 20091def test_http_with_credentials_fail_url(http_server_with_credentials):92    r = requests.get(93        url = 'http://localhost:{}/fail'.format(HTTP_PORT),94        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)95    )96    assert r.status_code in ERROR_RANGE97def test_https_with_credentials(https_server_with_credentials):98    r = requests.get(99        url = 'https://localhost:{}'.format(HTTPS_PORT),100        verify=SERVER_CA,101        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)102    )103    assert r.status_code == 200104def test_https_with_credentials_ignore_client_cert(https_server_with_credentials):105    r = requests.get(106        url = 'https://localhost:{}'.format(HTTPS_PORT),107        cert=(CLIENT_CRT, CLIENT_KEY),108        verify=SERVER_CA,109        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)110    )111    assert r.status_code == 200112def test_https_with_mutual_auth_with_credentials(https_server_with_mutual_auth_with_credentials):113    r = requests.get(114        url = 'https://localhost:{}'.format(HTTPSM_PORT),115        cert=(CLIENT_CRT, CLIENT_KEY),116        verify=SERVER_CA,117        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)118    )119    assert r.status_code == 200120def test_https_with_mutual_auth_with_credentials_wrong_password(https_server_with_mutual_auth_with_credentials):121    r = requests.get(122        url = 'https://localhost:{}'.format(HTTPSM_PORT),123        cert=(CLIENT_CRT, CLIENT_KEY),124        verify=SERVER_CA,125        auth=requests.auth.HTTPBasicAuth(USERNAME, '')126    )127    assert r.status_code == 401128def test_https_with_mutual_auth_with_credentials_ssl_failure(https_server_with_mutual_auth_with_credentials):129    with pytest.raises(requests.exceptions.SSLError):130        r = requests.get(131            url = 'https://localhost:{}'.format(HTTPSM_PORT),132            verify=SERVER_CA,133            auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)134        )135def test_mgmt_clear(mgmt_server, https_server_with_credentials):136    r = requests.get(137        url = 'https://localhost:{}'.format(HTTPS_PORT),138        verify=SERVER_CA,139        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)140    )141    r = requests.get(142        url = 'http://localhost:{}/clear'.format(MGMT_PORT),143        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)144    )145    r = requests.get(146        url = 'http://localhost:{}/stats'.format(MGMT_PORT),147        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)148    )149    assert r.content == b'{}'150def test_mgmt_stats(mgmt_server, https_server_with_credentials, https_server_with_mutual_auth_with_credentials):151    r = requests.get(152        url = 'http://localhost:{}/clear'.format(MGMT_PORT),153        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)154    )155    r = requests.get(156        url = 'https://localhost:{}'.format(HTTPS_PORT),157        verify=SERVER_CA,158        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)159    )160    r = requests.get(161        url = 'https://localhost:{}'.format(HTTPSM_PORT),162        cert=(CLIENT_CRT, CLIENT_KEY),163        verify=SERVER_CA,164        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)165    )166    r = requests.get(167        url = 'https://localhost:{}/fail'.format(HTTPSM_PORT),168        cert=(CLIENT_CRT, CLIENT_KEY),169        verify=SERVER_CA,170        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)171    )172    r = requests.get(173        url = 'http://localhost:{}/stats'.format(MGMT_PORT),174        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)175    )176    assert r.content == b'{"/": 2, "/fail": 1}'177def test_mgmt_history(mgmt_server, https_server_with_credentials):178    r = requests.get(179        url = 'http://localhost:{}/clear'.format(MGMT_PORT),180        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)181    )182    r = requests.get(183        url = 'https://localhost:{}'.format(HTTPS_PORT),184        verify=SERVER_CA,185        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)186    )187    r = requests.get(188        url = 'http://localhost:{}/history'.format(MGMT_PORT),189        auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)190    )191    content = json.loads(r.content)192    assert len(content) == 1193    assert content[0]["path"] == '/'...ZenodoTest.py
Source:ZenodoTest.py  
1import os2from os.path import join as p3from unittest.mock import Mock4import shutil5import pytest6from owmeta_core.datasource_loader import LoadFailed7from owmeta_movement.zenodo import ZenodoRecordDirLoader8import requests9def test_can_load(zenodo_dir_loader):10    '''11    Test that we provide for a matching record ID with no file name12    '''13    cut, https_server = zenodo_dir_loader14    os.mkdir(p(https_server.base_directory, 'record'))15    shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),16                    p(https_server.base_directory, 'record', '4074963'))17    ob = Mock()18    ob.zenodo_base_url.return_value = https_server.url19    ob.zenodo_id.return_value = 407496320    ob.zenodo_file_name.return_value = None21    assert cut.can_load(ob)22def test_can_load_no_file_attr(zenodo_dir_loader):23    '''24    Test that we provide for a matching record ID with no file name25    '''26    cut, https_server = zenodo_dir_loader27    os.mkdir(p(https_server.base_directory, 'record'))28    shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),29                    p(https_server.base_directory, 'record', '4074963'))30    ob = Mock()31    ob.zenodo_base_url.return_value = https_server.url32    ob.zenodo_id.return_value = 407496333    ob.zenodo_file_name.side_effect = AttributeError34    assert cut.can_load(ob)35def test_can_load_fail_not_found(zenodo_dir_loader):36    '''37    Test that we provide for a matching record ID with no file name38    '''39    cut, https_server = zenodo_dir_loader40    ob = Mock()41    ob.zenodo_base_url.return_value = https_server.url42    ob.zenodo_id.return_value = 407496343    ob.zenodo_file_name.return_value = None44    assert not cut.can_load(ob)45def test_can_load_file_fail_not_found(zenodo_dir_loader):46    '''47    Test that we provide for a matching record ID with no file name48    '''49    cut, https_server = zenodo_dir_loader50    recorddir = p(https_server.base_directory, 'record')51    os.mkdir(recorddir)52    shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),53                    p(recorddir, '4074963'))54    ob = Mock()55    ob.zenodo_base_url.return_value = https_server.url56    ob.zenodo_id.return_value = 407496357    ob.zenodo_file_name.return_value = 'CeMEE_MWT_MA.tar.gz'58    assert not cut.can_load(ob)59def test_can_load_file(zenodo_dir_loader):60    '''61    Test that we provide for a matching record ID with no file name62    '''63    cut, https_server = zenodo_dir_loader64    filesdir = p(https_server.base_directory, 'record', '4074963', 'files')65    os.makedirs(filesdir)66    file_name = p(filesdir, 'CeMEE_MWT_MA.tar.gz')67    with open(file_name, 'w') as f:68        f.write('blah')69    ob = Mock()70    ob.zenodo_base_url.return_value = https_server.url71    ob.zenodo_id.return_value = 407496372    ob.zenodo_file_name.return_value = 'CeMEE_MWT_MA.tar.gz'73    assert cut.can_load(ob)74def test_load_file(zenodo_dir_loader):75    cut, https_server = zenodo_dir_loader76    filesdir = p(https_server.base_directory, 'record', '4074963', 'files')77    os.makedirs(filesdir)78    target_file_name = 'CeMEE_MWT_MA.tar.gz'79    file_name = p(filesdir, target_file_name)80    with open(file_name, 'w') as f:81        f.write('blah')82    ob = Mock()83    ob.zenodo_base_url.return_value = https_server.url84    ob.zenodo_id.return_value = 407496385    ob.zenodo_file_name.return_value = target_file_name86    dsdir = cut.load(ob)87    with open(p(dsdir, target_file_name), 'r') as f:88        assert f.read() == 'blah'89def test_load_some_files(tmp_path, https_server):90    '''91    Loads any files available--no error is thrown if some are absent.92    '''93    recorddir = p(https_server.base_directory, 'record')94    os.mkdir(recorddir)95    shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),96                    p(recorddir, '4074963.html'))97    def handler(server_data):98        class handler_class(server_data.basic_handler):99            def do_GET(self):100                if self.path.endswith('4074963'):101                    self.handle_request(200)102                    with open(p(recorddir, '4074963.html'), 'rb') as f:103                        shutil.copyfileobj(f, self.wfile)104                else:105                    super().do_GET()106        return handler_class107    https_server.make_server(handler)108    https_server.restart()109    session = requests.Session()110    https_server.trust_server(session)111    cut = ZenodoRecordDirLoader(tmp_path, lambda: session)112    filesdir = p(https_server.base_directory, 'record', '4074963', 'files')113    os.makedirs(filesdir)114    target_file_name = 'CeMEE_MWT_MA.tar.gz'115    file_name = p(filesdir, target_file_name)116    with open(file_name, 'w') as f:117        f.write('blah')118    ob = Mock()119    ob.zenodo_base_url.return_value = https_server.url120    ob.zenodo_id.return_value = 4074963121    ob.zenodo_file_name.return_value = None122    with pytest.raises(LoadFailed):123        cut.load(ob)124@pytest.mark.inttest125def test_load_real():126    '''127    Test actually loading from Zenodo128    '''129@pytest.fixture130def zenodo_dir_loader(tmp_path, https_server):131    session = requests.Session()132    https_server.trust_server(session)...service.py
Source:service.py  
1"""Network service to receive requests"""2import getopt3import logging4import multiprocessing5import os6import ssl7import sys8from .handler import handler9from http.server import HTTPServer10from socketserver import ThreadingMixIn11from .tcp.service import Service as TcpService12class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):13    """Handle requests in a separate thread."""14class HttpsService(multiprocessing.Process):15    """Handles HTTPS protocol network activity"""16    def __init__(self, key_folder, port, params, server_class=ThreadedHTTPServer):17        super().__init__()18        logging.info(f'SSL with {key_folder + "/key.pem"} and {key_folder + "/cert.pem"} is used')19        self.https_server = server_class(('', port), handler(params))20        self.https_server.socket = ssl.wrap_socket(self.https_server.socket,21                                                   keyfile=key_folder + "/key.pem",22                                                   certfile=key_folder + '/cert.pem',23                                                   server_side=True)24    def run(self) -> None:25        """Starts service"""26        try:27            self.https_server.serve_forever()28        except KeyboardInterrupt:29            self.https_server.server_close()30    def join(self, timeout=None) -> None:31        """Implements service thread-safe stop"""32        if super().is_alive():33            self.https_server.server_close()34            super().join(timeout)35class Service:36    """Program launcher. Analyses terminal options and starts http server"""37    @staticmethod38    def print_options():39        """Informs about program terminal arguments"""40        print("params:\n\t-p(--ports) ports to bind[http,https,rtsp] (def 4555,4556,4557)\n\t"41              "-r(--root) files directory(req)\n\t"42              "-s(--segment) segment duration floor\n\t"43              "-c(--cache) cache segmentation as .*.cache files\n\t"44              "-b(--basic) user:password@realm (use Basic Authorization)\n\t"45              "-d(--digest) user:password@realm (use Digest Authorization)\n\t"46              "-k(--keys) directory with key.pem and cert.pem files (req. for https)\n\t"47              "-v(--verb) be verbose\n\t"48              "-h(--help) this help")49    def __init__(self):50        self.segment_makers = {}51    def run(self, ports, params, server_class=ThreadedHTTPServer):52        """Starts http server"""53        logging.basicConfig(level=logging.INFO)54        params['segment_makers'] = self.segment_makers55        tcp_server = TcpService(('', ports[2]), params)56        http_server = server_class(('', ports[0]), handler(params))57        ssl_key_folder = params.get('keys')58        https_server = None59        if ssl_key_folder and os.path.isfile(ssl_key_folder+'/key.pem') and os.path.isfile(ssl_key_folder+'/cert.pem'):60            https_server = HttpsService(ssl_key_folder, ports[1], params)61        logging.info('Starting...')62        try:63            tcp_server.start()64            if https_server:65                https_server.start()66            http_server.serve_forever()67        except KeyboardInterrupt:68            pass69        http_server.server_close()70        https_server and https_server.join()71        tcp_server.join()72        logging.info('Stopping')73def start():74    """Program start point"""75    argv = sys.argv[1:]76    try:77        opts, args = getopt.getopt(argv,78                                   "hp:r:s:b:d:ck:v",79                                   ["help",80                                    "ports=",81                                    "root=",82                                    "segment=",83                                    "basic=",84                                    "digest=",85                                    "cache",86                                    "keys=",87                                    "verb"])88        if args:89            Service.print_options()90            sys.exit()91    except getopt.GetoptError as error:92        print(error)93        sys.exit()94    ports = [4555, 4556, 4557]95    params = {}96    try:97        for opt, arg in opts:98            if opt in ('-h', '--help'):99                Service.print_options()100                sys.exit()101            elif opt in ('-p', '--ports'):102                ports = [int(k) for k in arg.split(',')]103                if len(ports) != 3:104                    Service.print_options()105                    sys.exit()106            elif opt in ('-r', '--root'):107                params['root'] = arg108            elif opt in ('-s', '--segment'):109                params['segment'] = float(arg)110            elif opt in ('-b', '--basic'):111                params['basic'] = arg112            elif opt in ('-d', '--digest'):113                params['digest'] = arg114            elif opt in ('-c', '--cache'):115                params['cache'] = True116            elif opt in ('-k', '-keys'):117                params['keys'] = arg118            elif opt in ('-v', '--verb'):119                params['verb'] = True120    except ValueError as error:121        print(error)122        Service.print_options()123        sys.exit()...LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!
