eggfetch-python 0.1.4

Python sync and asyncio bindings for the eggfetch HTTP engine (Rust core via PyO3; Python users install from PyPI)
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

"""Interpreter shutdown tests.

These tests verify that the Python interpreter shuts down cleanly even
when clients are not explicitly closed, when streaming responses are
abandoned, or when multiple clients are created and destroyed rapidly.
Each test runs a subprocess to verify no deadlocks, panics, leaked
threads, or excessive shutdown delay.
"""

import subprocess
import sys
import textwrap
import time

import pytest


PYTHON = sys.executable


def run_subprocess(script: str, timeout: int = 10) -> subprocess.CompletedProcess:
    """Run a Python script as a subprocess and return the result."""
    return subprocess.run(
        [PYTHON, "-c", textwrap.dedent(script)],
        capture_output=True,
        text=True,
        timeout=timeout,
    )


# ---------------------------------------------------------------------------
# Client not explicitly closed
# ---------------------------------------------------------------------------


class TestInterpreterShutdown:
    """Verify clean interpreter shutdown under various conditions."""

    def test_client_not_closed(self):
        """Client not explicitly closed should not deadlock on exit."""
        result = run_subprocess("""
            import eggfetch
            client = eggfetch.Client()
            # Do not call client.close()
        """)
        assert result.returncode == 0, (
            f"Non-zero exit: rc={result.returncode}\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )

    def test_async_client_not_closed(self):
        """AsyncClient not explicitly closed should not deadlock on exit."""
        result = run_subprocess("""
            import asyncio, eggfetch
            async def main():
                client = eggfetch.AsyncClient()
                # Do not call client.close()
            asyncio.run(main())
        """)
        assert result.returncode == 0, (
            f"Non-zero exit: rc={result.returncode}\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )

    def test_context_manager_exit(self):
        """Context manager should close client cleanly."""
        result = run_subprocess("""
            import eggfetch
            with eggfetch.Client() as client:
                pass
        """)
        assert result.returncode == 0, (
            f"Non-zero exit: rc={result.returncode}\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )

    def test_repeated_construction_and_destruction(self):
        """Repeatedly create and destroy clients should not leak resources."""
        result = run_subprocess("""
            import eggfetch
            for _ in range(10):
                client = eggfetch.Client()
                client.close()
        """)
        assert result.returncode == 0, (
            f"Non-zero exit: rc={result.returncode}\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )

    def test_multiple_clients_not_closed(self):
        """Multiple clients not closed should not deadlock on exit."""
        result = run_subprocess("""
            import eggfetch
            clients = [eggfetch.Client() for _ in range(5)]
            # Do not close any of them
        """)
        assert result.returncode == 0, (
            f"Non-zero exit: rc={result.returncode}\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )

    def test_context_manager_with_exception(self):
        """Context manager should close client even if exception occurs."""
        result = run_subprocess("""
            import eggfetch
            try:
                with eggfetch.Client() as client:
                    raise ValueError("test exception")
            except ValueError:
                pass
        """)
        assert result.returncode == 0, (
            f"Non-zero exit: rc={result.returncode}\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )

    def test_client_outlives_scope(self):
        """Client used after context manager exit should raise ValueError."""
        result = run_subprocess("""
            import eggfetch
            with eggfetch.Client() as client:
                pass
            try:
                client.get("http://example.com")
                print("ERROR: should have raised")
            except ValueError as e:
                if "closed" in str(e):
                    print("OK: got expected ValueError")
                else:
                    print(f"ERROR: unexpected ValueError: {e}")
        """)
        assert result.returncode == 0, (
            f"Non-zero exit: rc={result.returncode}\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )
        assert "OK: got expected ValueError" in result.stdout