xref: /curl/tests/http/testenv/client.py (revision f0c446ab) 
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#
27import pytest
28import json
29import logging
30import os
31import re
32import shutil
33import subprocess
34from datetime import timedelta, datetime
35from typing import List, Optional, Dict, Union
36from urllib.parse import urlparse
37
38from . import ExecResult
39from .env import Env
40
41
42log = logging.getLogger(__name__)
43
44
45class LocalClient:
46
47    def __init__(self, name: str, env: Env, run_dir: Optional[str] = None,
48                 timeout: Optional[float] = None,
49                 run_env: Optional[Dict[str,str]] = None):
50        self.name = name
51        self.path = os.path.join(env.project_dir, f'tests/http/clients/{name}')
52        self.env = env
53        self._run_env= run_env
54        self._timeout = timeout if timeout else env.test_timeout
55        self._curl = os.environ['CURL'] if 'CURL' in os.environ else env.curl
56        self._run_dir = run_dir if run_dir else os.path.join(env.gen_dir, name)
57        self._stdoutfile = f'{self._run_dir}/stdout'
58        self._stderrfile = f'{self._run_dir}/stderr'
59        self._rmrf(self._run_dir)
60        self._mkpath(self._run_dir)
61
62    @property
63    def run_dir(self) -> str:
64        return self._run_dir
65
66    @property
67    def stderr_file(self) -> str:
68        return self._stderrfile
69
70    def exists(self) -> bool:
71        return os.path.exists(self.path)
72
73    def download_file(self, i: int) -> str:
74        return os.path.join(self._run_dir, f'download_{i}.data')
75
76    def _rmf(self, path):
77        if os.path.exists(path):
78            return os.remove(path)
79
80    def _rmrf(self, path):
81        if os.path.exists(path):
82            return shutil.rmtree(path)
83
84    def _mkpath(self, path):
85        if not os.path.exists(path):
86            return os.makedirs(path)
87
88    def run(self, args):
89        self._rmf(self._stdoutfile)
90        self._rmf(self._stderrfile)
91        start = datetime.now()
92        exception = None
93        myargs = [self.path]
94        myargs.extend(args)
95        try:
96            with open(self._stdoutfile, 'w') as cout:
97                with open(self._stderrfile, 'w') as cerr:
98                    p = subprocess.run(myargs, stderr=cerr, stdout=cout,
99                                       cwd=self._run_dir, shell=False,
100                                       input=None, env=self._run_env,
101                                       timeout=self._timeout)
102                    exitcode = p.returncode
103        except subprocess.TimeoutExpired:
104            log.warning(f'Timeout after {self._timeout}s: {args}')
105            exitcode = -1
106            exception = 'TimeoutExpired'
107        coutput = open(self._stdoutfile).readlines()
108        cerrput = open(self._stderrfile).readlines()
109        return ExecResult(args=myargs, exit_code=exitcode, exception=exception,
110                          stdout=coutput, stderr=cerrput,
111                          duration=datetime.now() - start)
112
113    def dump_logs(self):
114        lines = []
115        lines.append('>>--stdout ----------------------------------------------\n')
116        lines.extend(open(self._stdoutfile).readlines())
117        lines.append('>>--stderr ----------------------------------------------\n')
118        lines.extend(open(self._stderrfile).readlines())
119        lines.append('<<-------------------------------------------------------\n')
120        return ''.join(lines)
121