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
|
# SPDX-License-Identifier: MIT OR Apache-2.0
# This file is dual licensed under the terms of the Apache License, Version
# 2.0, and the MIT License. See the LICENSE file in the root of this
# repository for complete details.
import multiprocessing
import sys
import pytest
from structlog._utils import get_processname
class TestGetProcessname:
def test_default(self):
"""
The returned process name matches the name of the current process from
the `multiprocessing` module.
"""
assert get_processname() == multiprocessing.current_process().name
def test_changed(self, monkeypatch: pytest.MonkeyPatch):
"""
The returned process name matches the name of the current process from
the `multiprocessing` module if it is not the default.
"""
tmp_name = "fakename"
monkeypatch.setattr(
target=multiprocessing.current_process(),
name="name",
value=tmp_name,
)
assert get_processname() == tmp_name
def test_no_multiprocessing(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""
The returned process name is the default process name if the
`multiprocessing` module is not available.
"""
tmp_name = "fakename"
monkeypatch.setattr(
target=multiprocessing.current_process(),
name="name",
value=tmp_name,
)
monkeypatch.setattr(
target=sys,
name="modules",
value={},
)
assert get_processname() == "n/a"
def test_exception(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""
The returned process name is the default process name when an exception
is thrown when an attempt is made to retrieve the current process name
from the `multiprocessing` module.
"""
def _current_process() -> None:
raise RuntimeError("test")
monkeypatch.setattr(
target=multiprocessing,
name="current_process",
value=_current_process,
)
assert get_processname() == "n/a"
|
