File: test_utils.py

package info (click to toggle)
python-tuf 6.0.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,300 kB
  • sloc: python: 7,738; makefile: 8
file content (79 lines) | stat: -rw-r--r-- 2,163 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
# Copyright 2020, TUF contributors
# SPDX-License-Identifier: MIT OR Apache-2.0

"""
<Program Name>
  test_utils.py

<Author>
  Martin Vrachev.

<Started>
  October 21, 2020.

<Copyright>
  See LICENSE-MIT OR LICENSE for licensing information.

<Purpose>
  Provide tests for some of the functions in utils.py module.
"""

import logging
import os
import socket
import sys
import unittest

from tests import utils

logger = logging.getLogger(__name__)


def can_connect(port: int) -> bool:
    """Check if a socket can connect on the given port"""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect(("localhost", port))
        return True
    except Exception:  # noqa: BLE001
        return False
    finally:
        # The process will always enter in finally even after return.
        if sock:
            sock.close()


class TestServerProcess(unittest.TestCase):
    """Test functionality provided in TestServerProcess from tests/utils.py."""

    def test_simple_server_startup(self) -> None:
        # Test normal case
        server_process_handler = utils.TestServerProcess(log=logger)

        # Make sure we can connect to the server
        self.assertTrue(can_connect(server_process_handler.port))
        server_process_handler.clean()

    def test_cleanup(self) -> None:
        # Test normal case
        server_process_handler = utils.TestServerProcess(
            log=logger, server=os.path.join(utils.TESTS_DIR, "simple_server.py")
        )

        server_process_handler.clean()

        # Check if the process has successfully been killed.
        self.assertFalse(server_process_handler.is_process_running())

    def test_server_exit_before_timeout(self) -> None:
        with self.assertRaises(utils.TestServerProcessError):
            utils.TestServerProcess(logger, server="non_existing_server.py")

        # Test starting a server which immediately exits."
        with self.assertRaises(utils.TestServerProcessError):
            utils.TestServerProcess(logger, server="fast_server_exit.py")


if __name__ == "__main__":
    utils.configure_test_logging(sys.argv)
    unittest.main()