Coverage for tests/integrations/conftest.py: 75%

1from __future__ import annotations 

2 

3import sys 

4import shutil 

5import subprocess 

6from typing import TYPE_CHECKING, Any, Union, Iterator, Optional, cast 

7from pathlib import Path 

8from functools import lru_cache 

9 

10import pytest 

11from _pytest._code import ExceptionInfo 

12from _pytest.nodes import Node 

13from _pytest.config import Config 

14from _pytest._code.code import TerminalRepr 

15 

16if TYPE_CHECKING: 

17 from _pytest._code.code import _TracebackStyle 

18 

19 

20ROOTDIR = Path(__file__).parent.parent.parent 

21 

22 

23class IntegrationError(AssertionError): 

24 def __init__(self, message: Optional[str] = None, lineno: int = 0) -> None: # pragma: no cover 

25 self.message = message or '' 

26 self.lineno = lineno 

27 super().__init__() 

28 

29 def __str__(self) -> str: # pragma: no cover 

30 return self.message 

31 

32 

33def resolve_path(path: Path) -> Path: 

34 return path.relative_to(Path.cwd()) 

35 

36 

37def is_integration_test_file(path: Path) -> bool: 

38 # disable integration tests as they are currently broken 

39 return False 

40 

41 

42def pytest_ignore_collect(collection_path: Path, config: Config) -> Optional[bool]: 

43 """We need to ignore any integration test sub-paths 

44 

45 For example we need to include 

46 

47 tests/integrations 

48 tests/integrations/* 

49 tests/integrations/basic/test.sh 

50 

51 but ignore anything that looks like 

52 

53 tests/integrations/basic/* 

54 tests/integrations/basic/conftest.py 

55 tests/integrations/basic/tests/test_foo.py 

56 """ 

57 path = resolve_path(collection_path) 

58 if path.parts[:2] != ('tests', 'integrations'): # pragma: no cover 

59 # not an integration test 

60 return None 

61 

62 if len(path.parts) <= 3: 

63 # integration root dir, leave handling to pytest 

64 return None 

65 

66 return path.parts[-1] != 'test.sh' 

67 

68 

69def pytest_collect_file(file_path: Path, parent: Node) -> Optional['IntegrationTestFile']: 

70 path = resolve_path(file_path) 

71 if path.suffix == '.sh' and is_integration_test_file(path) and sys.platform != 'win32': 71 ↛ 72line 71 didn't jump to line 72, because the condition on line 71 was never true

72 return IntegrationTestFile.from_parent(parent, path=path) 

73 

74 return None 

75 

76 

77@lru_cache(maxsize=None) 

78def create_wheels() -> None: 

79 dist = ROOTDIR / '.tests_cache' / 'dist' 

80 if dist.exists(): # pragma: no cover 

81 shutil.rmtree(str(dist)) 

82 

83 result = subprocess.run( 

84 [ 

85 sys.executable, 

86 'setup.py', 

87 'bdist_wheel', 

88 '--dist-dir=.tests_cache/dist', 

89 ], 

90 check=False, 

91 stdout=subprocess.PIPE, 

92 stderr=subprocess.STDOUT, 

93 cwd=str(ROOTDIR), 

94 ) 

95 if result.returncode != 0: # pragma: no cover 

96 print(result.stdout.decode('utf-8')) 

97 raise RuntimeError('Could not build wheels, see output above.') 

98 

99 

100class IntegrationTestItem(pytest.Item): 

101 def __init__( 

102 self, 

103 name: str, 

104 parent: Optional['IntegrationTestFile'] = None, 

105 config: Optional[Config] = None, 

106 *, 

107 path: Path, 

108 ) -> None: 

109 super().__init__(name, parent, config) 

110 self.path = path 

111 self.starting_lineno = 1 

112 

113 def setup(self) -> None: 

114 create_wheels() 

115 

116 def runtest(self) -> None: 

117 # TODO: include exit code in pytest failure short description 

118 result = subprocess.run( 

119 [str(self.path)], 

120 cwd=str(self.path.parent), 

121 stdout=subprocess.PIPE, 

122 stderr=subprocess.STDOUT, 

123 check=False, 

124 ) 

125 print(result.stdout.decode('utf-8')) 

126 if result.returncode != 0: # pragma: no cover 

127 raise IntegrationError(f'Executing `{self.path}` returned non-zero exit code {result.returncode}') 

128 

129 def repr_failure( 

130 self, 

131 excinfo: ExceptionInfo[BaseException], 

132 style: Optional['_TracebackStyle'] = None, 

133 ) -> Union[str, TerminalRepr]: # pragma: no cover 

134 if isinstance(excinfo.value, IntegrationError): 

135 return f'IntegrationError: {excinfo.value.message}' 

136 

137 return super().repr_failure(excinfo, style=style) 

138 

139 def reportinfo(self): 

140 return self.fspath, 0, f'integration: {self.name}' 

141 

142 

143class IntegrationTestFile(pytest.File): 

144 @classmethod 

145 def from_parent(cls, *args: Any, **kwargs: Any) -> 'IntegrationTestFile': 

146 return cast( 

147 IntegrationTestFile, 

148 super().from_parent(*args, **kwargs), # type: ignore[no-untyped-call] 

149 ) 

150 

151 def collect(self) -> Iterator[IntegrationTestItem]: 

152 path = Path(self.fspath) 

153 yield IntegrationTestItem.from_parent(parent=self, name=path.parent.name, path=path)