Coverage for tests/test_cli/test_dev.py: 100%

1import re 

2import sys 

3import time 

4import signal 

5import threading 

6import subprocess 

7from typing import List 

8 

9import pytest 

10from _pytest.monkeypatch import MonkeyPatch 

11 

12from prisma.http import client 

13from prisma.utils import temp_env_update 

14from prisma.cli.commands import dev 

15 

16from ..utils import Runner, Testdir 

17 

18PRISMA_MODULE = ['coverage', 'run', '-m', 'prisma'] 

19 

20 

21def test_playground_skip_generate_no_client(runner: Runner, monkeypatch: MonkeyPatch) -> None: 

22 """Passing --skip-generate with a non-generated client raises an error""" 

23 

24 def mock_return(mod: str) -> bool: 

25 return False 

26 

27 monkeypatch.setattr(dev, 'module_exists', mock_return, raising=True) 

28 result = runner.invoke(['py', 'dev', 'playground', '--skip-generate']) 

29 assert result.exit_code == 1 

30 assert result.output == 'Prisma Client Python has not been generated yet.\n' 

31 

32 

33@pytest.mark.asyncio 

34@pytest.mark.skip( 

35 reason='Test is currently broken and the feature is not intended for users anyway', 

36) 

37async def test_playground(testdir: Testdir) -> None: 

38 """Starts local HTTP server""" 

39 

40 def output_reader(proc: 'subprocess.Popen[bytes]', lines: List[str]) -> List[str]: 

41 assert proc.stdout is not None 

42 for line in iter(proc.stdout.readline, b''): 

43 lines.append(line.decode('utf-8')) 

44 

45 return lines 

46 

47 lines: List[str] = [] 

48 schema = testdir.make_schema() 

49 args = PRISMA_MODULE + ['py', 'dev', 'playground', f'--schema={schema}'] 

50 

51 # ensure host information is still logged 

52 # with DEBUG logging disabled 

53 with temp_env_update({'PRISMA_PY_DEBUG': '0'}): 

54 process = subprocess.Popen(args=args, stdout=subprocess.PIPE) 

55 

56 thread = threading.Thread(target=output_reader, args=(process, lines)) 

57 

58 try: 

59 thread.start() 

60 

61 # TODO: don't naively sleep, instead check the output every <x> time 

62 # until it is what we want or a timeout is reached 

63 time.sleep(8) 

64 

65 stdout = ''.join(lines) 

66 print(stdout) 

67 

68 assert 'Generated Prisma Client Python' in stdout 

69 

70 match = re.search(r'Started http server on (?P<url>http://127.0.0.1:\d+)', stdout) 

71 assert match is not None 

72 url = match.group('url') 

73 

74 resp = await client.request('GET', url) 

75 assert resp.status == 200 

76 assert '<title>Rust Playground</title>' in await resp.text() 

77 finally: 

78 if sys.platform == 'win32': # pragma: no cover 

79 sig = signal.CTRL_C_EVENT 

80 else: 

81 sig = signal.SIGINT 

82 

83 process.send_signal(sig) 

84 thread.join(timeout=5)