1import asyncio
2import json
3import os
4import shutil
5
6import pytest
7from tornado.httpclient import HTTPClientError
8from traitlets.config import Config
9
10
11@pytest.fixture
12def terminal_path(tmp_path):
13    subdir = tmp_path.joinpath("terminal_path")
14    subdir.mkdir()
15
16    yield subdir
17
18    shutil.rmtree(str(subdir), ignore_errors=True)
19
20
21CULL_TIMEOUT = 10
22CULL_INTERVAL = 3
23
24
25@pytest.fixture
26def jp_server_config():
27    return Config(
28        {
29            "ServerApp": {
30                "TerminalManager": {
31                    "cull_inactive_timeout": CULL_TIMEOUT,
32                    "cull_interval": CULL_INTERVAL,
33                }
34            }
35        }
36    )
37
38
39async def test_no_terminals(jp_fetch):
40    resp_list = await jp_fetch(
41        "api",
42        "terminals",
43        method="GET",
44        allow_nonstandard_methods=True,
45    )
46
47    data = json.loads(resp_list.body.decode())
48
49    assert len(data) == 0
50
51
52async def test_terminal_create(jp_fetch, jp_cleanup_subprocesses):
53    resp = await jp_fetch(
54        "api",
55        "terminals",
56        method="POST",
57        allow_nonstandard_methods=True,
58    )
59    term = json.loads(resp.body.decode())
60    assert term["name"] == "1"
61
62    resp_list = await jp_fetch(
63        "api",
64        "terminals",
65        method="GET",
66        allow_nonstandard_methods=True,
67    )
68
69    data = json.loads(resp_list.body.decode())
70
71    assert len(data) == 1
72    assert data[0] == term
73    await jp_cleanup_subprocesses()
74
75
76async def test_terminal_create_with_kwargs(
77    jp_fetch, jp_ws_fetch, terminal_path, jp_cleanup_subprocesses
78):
79    resp_create = await jp_fetch(
80        "api",
81        "terminals",
82        method="POST",
83        body=json.dumps({"cwd": str(terminal_path)}),
84        allow_nonstandard_methods=True,
85    )
86
87    data = json.loads(resp_create.body.decode())
88    term_name = data["name"]
89
90    resp_get = await jp_fetch(
91        "api",
92        "terminals",
93        term_name,
94        method="GET",
95        allow_nonstandard_methods=True,
96    )
97
98    data = json.loads(resp_get.body.decode())
99
100    assert data["name"] == term_name
101    await jp_cleanup_subprocesses()
102
103
104async def test_terminal_create_with_cwd(
105    jp_fetch, jp_ws_fetch, terminal_path, jp_cleanup_subprocesses
106):
107    resp = await jp_fetch(
108        "api",
109        "terminals",
110        method="POST",
111        body=json.dumps({"cwd": str(terminal_path)}),
112        allow_nonstandard_methods=True,
113    )
114
115    data = json.loads(resp.body.decode())
116    term_name = data["name"]
117
118    ws = await jp_ws_fetch("terminals", "websocket", term_name)
119
120    ws.write_message(json.dumps(["stdin", "pwd\r\n"]))
121
122    message_stdout = ""
123    while True:
124        try:
125            message = await asyncio.wait_for(ws.read_message(), timeout=5.0)
126        except asyncio.TimeoutError:
127            break
128
129        message = json.loads(message)
130
131        if message[0] == "stdout":
132            message_stdout += message[1]
133
134    ws.close()
135
136    assert os.path.basename(terminal_path) in message_stdout
137    await jp_cleanup_subprocesses()
138
139
140async def test_culling_config(jp_server_config, jp_configurable_serverapp):
141    terminal_mgr_config = jp_configurable_serverapp().config.ServerApp.TerminalManager
142    assert terminal_mgr_config.cull_inactive_timeout == CULL_TIMEOUT
143    assert terminal_mgr_config.cull_interval == CULL_INTERVAL
144    terminal_mgr_settings = jp_configurable_serverapp().web_app.settings["terminal_manager"]
145    assert terminal_mgr_settings.cull_inactive_timeout == CULL_TIMEOUT
146    assert terminal_mgr_settings.cull_interval == CULL_INTERVAL
147
148
149async def test_culling(jp_server_config, jp_fetch, jp_cleanup_subprocesses):
150    # POST request
151    resp = await jp_fetch(
152        "api",
153        "terminals",
154        method="POST",
155        allow_nonstandard_methods=True,
156    )
157    term = json.loads(resp.body.decode())
158    term_1 = term["name"]
159    last_activity = term["last_activity"]
160
161    culled = False
162    for i in range(CULL_TIMEOUT + CULL_INTERVAL):
163        try:
164            resp = await jp_fetch(
165                "api",
166                "terminals",
167                term_1,
168                method="GET",
169                allow_nonstandard_methods=True,
170            )
171        except HTTPClientError as e:
172            assert e.code == 404
173            culled = True
174            break
175        else:
176            await asyncio.sleep(1)
177
178    assert culled
179    await jp_cleanup_subprocesses()
180