Coverage for src/robotide/controller/dataloader.py: 74%

1# Copyright 2008-2015 Nokia Networks 

2# Copyright 2016- Robot Framework Foundation 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15 

16import os 2 ctx1ab

17from threading import Thread 2 ctx1ab

18 

19from robotide.lib.compat.parsing import language as lang 2 ctx1ab

20from robotide.lib.robot.errors import DataError 2 ctx1ab

21from .. import robotapi 2 ctx1ab

22 

23 

24class DataLoader(object): 2 ctx1ab

25 

26 def __init__(self, namespace, settings, language=None): 2 ctx1ab

27 self.namespace = namespace 16 ctx1adxyzABCDEFGHsef

28 self.namespace.reset_resource_and_library_cache() 16 ctx1adxyzABCDEFGHsef

29 self._settings = settings 16 ctx1adxyzABCDEFGHsef

30 self.language = language 16 ctx1adxyzABCDEFGHsef

31 

32 def load_datafile(self, path, load_observer, language=None): 2 ctx1ab

33 if not language: 17 ctx1ahdijkulmnopqrsef

34 self.language = lang.check_file_language(path) 16 ctx1ahdijkulmnopqref

35 else: 

36 self.language = language 2 ctx1as

37 return self._load(_DataLoader(path, self._settings, language=self.language), load_observer) 17 ctx1ahdijkulmnopqrsef

38 

39 def load_initfile(self, path, load_observer, language=None): 2 ctx1ab

40 if not language: 

41 self.language = lang.check_file_language(path) 

42 else: 

43 self.language = language 

44 # print(f"DEBUG: datloder.py DataLoader.load_initfile ENTER self.language={self.language} language={language}") 

45 return self._load(_InitFileLoader(path, language=self.language), load_observer) 

46 

47 def load_resource_file(self, datafile, load_observer, language=None): 2 ctx1ab

48 if not language: 48 ↛ 51line 48 didn't jump to line 51 because the condition on line 48 was always true4 ctx1avtw

49 self.language = lang.check_file_language(datafile) 4 ctx1avtw

50 else: 

51 self.language = language 

52 return self._load(_ResourceLoader(datafile, self.namespace.get_resource, language=self.language), load_observer) 4 ctx1avtw

53 

54 def resources_for(self, datafile, load_observer, language=None): 2 ctx1ab

55 if not language: 55 ↛ 58line 55 didn't jump to line 58 because the condition on line 55 was always true(empty)

56 self.language = lang.check_file_language(datafile.source) (empty)

57 else: 

58 self.language = language 

59 return self._load(_ResourceLoader(datafile, self.namespace.get_resources, language=self.language), (empty)

60 load_observer) 

61 

62 def _load(self, loader, load_observer): 2 ctx1ab

63 self._wait_until_loaded(loader, load_observer) 20 ctx1avhdijkultmwnopqrsef

64 return loader.result 2 ctx1at

65 

66 @staticmethod 2 ctx1ab

67 def _wait_until_loaded(loader, load_observer): 2 ctx1ab

68 loader.start() 20 ctx1avhdijkultmwnopqrsef

69 load_observer.notify() 18 ctx1avhdijkltmwnopqref

70 while loader.is_alive(): 19 ctx1acvhdijkltmwnopqref

71 loader.join(0.1) 19 ctx1acvhdijkltmwnopqref

72 load_observer.notify() 3 ctx1act

73 

74 

75class _DataLoaderThread(Thread): 2 ctx1ab

76 

77 def __init__(self): 2 ctx1ab

78 Thread.__init__(self) 20 ctx1avhdijkultmwnopqrsef

79 self.result = None 20 ctx1avhdijkultmwnopqrsef

80 

81 def run(self): 2 ctx1ab

82 try: 20 ctx1avhdijkultmwnopqrsef

83 self.result = self._run() 20 ctx1avhdijkultmwnopqrsef

84 except Exception as e: 

85 print("DEBUG: exception at DataLoader %s\n" % str(e)) 

86 

87 def _run(self): 2 ctx1ab

88 return NotImplemented 

89 

90 

91class _DataLoader(_DataLoaderThread): 2 ctx1ab

92 

93 def __init__(self, path, settings, language=None): 2 ctx1ab

94 _DataLoaderThread.__init__(self) 17 ctx1ahdijkulmnopqrsef

95 self._path = path 17 ctx1ahdijkulmnopqrsef

96 self._settings = settings 17 ctx1ahdijkulmnopqrsef

97 self.language = language 17 ctx1ahdijkulmnopqrsef

98 

99 def _run(self): 2 ctx1ab

100 if not self.language: 17 ctx1ahdijkulmnopqrsef

101 try: 16 ctx1ahdijkulmnopqref

102 self.language = lang.check_file_language(self._path) 16 ctx1ahdijkulmnopqref

103 except Exception: 

104 self.language = 'en' 

105 return test_data(source=self._path, settings=self._settings, language=self.language) 17 ctx1ahdijkulmnopqrsef

106 

107 

108class _InitFileLoader(_DataLoaderThread): 2 ctx1ab

109 

110 def __init__(self, path, settings=None, language=None): 2 ctx1ab

111 _DataLoaderThread.__init__(self) 

112 self._path = path 

113 self._settings = settings 

114 self.language = language 

115 

116 def _run(self): 2 ctx1ab

117 result = robotapi.TestDataDirectory(source=os.path.dirname(self._path), settings=self._settings, 

118 language=self.language) 

119 result.initfile = self._path 

120 if not self.language: 

121 try: 

122 self.language = lang.check_file_language(self._path) 

123 except Exception: 

124 self.language = 'en' 

125 robotapi.FromFilePopulator(result, lang=self.language).populate(self._path) 

126 return result 

127 

128 

129class _ResourceLoader(_DataLoaderThread): 2 ctx1ab

130 

131 def __init__(self, datafile, resource_loader, language=None): 2 ctx1ab

132 _DataLoaderThread.__init__(self) 4 ctx1avtw

133 self._datafile = datafile 4 ctx1avtw

134 self.language = language 4 ctx1avtw

135 self._loader = resource_loader 4 ctx1avtw

136 

137 def _run(self): 2 ctx1ab

138 return self._loader(self._datafile, self.language) 4 ctx1avtw

139 

140 

141class TestDataDirectoryWithExcludes(robotapi.TestDataDirectory): 2 ctx1ab

142 

143 def __init__(self, parent, source, settings, language=None): 2 ctx1ab

144 self._settings = settings 2 ctx1cg

145 self.language = language 2 ctx1cg

146 robotapi.TestDataDirectory.__init__(self, parent, source, settings=self._settings, language=self.language) 2 ctx1cg

147 

148 def add_child(self, path, include_suites, extensions=None, 2 ctx1ab

149 warn_on_skipped=False, language=None): 

150 if not self._settings.excludes.contains(path): 150 ↛ 153line 150 didn't jump to line 153 because the condition on line 150 was always true2 ctx1cg

151 self.children.append(test_data(parent=self, source=path, settings=self._settings, language=self.language)) 2 ctx1cg

152 else: 

153 self.children.append(ExcludedDirectory(self, path, language=self.language)) 

154 

155 

156def test_data(source, parent=None, settings=None, language=None): 2 ctx1ab

157 """Parses a file or directory to a corresponding model object. 

158 

159 :param source: path where test data is read from. 

160 :returns: :class:`~.model.TestDataDirectory` if `source` is a directory, 

161 :class:`~.model.TestCaseFile` otherwise. 

162 """ 

163 # Check if opening an __init__.robot 

164 if os.path.basename(source) == '__init__.robot': 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true2 ctx1cg

165 source = os.path.dirname(source) 

166 if os.path.isdir(source): 2 ctx1cg

167 if not language: 167 ↛ 173line 167 didn't jump to line 173 because the condition on line 167 was always true2 ctx1cg

168 init_file = os.path.join(source, '__init__.robot') 2 ctx1cg

169 if os.path.isfile(init_file): 2 ctx1cg

170 language = lang.check_file_language(init_file) 1 ctx1c

171 # print(f"DEBUG: Dataloader TestCaseFile init file {init_file=}\n" 

172 # f" language={language} {source=}") 

173 data = TestDataDirectoryWithExcludes(parent, source, settings, language) 2 ctx1cg

174 # print("DEBUG: Dataloader testdata %s\n" % data.name) 

175 data.populate() 2 ctx1cg

176 # print("DEBUG: Dataloader after populate %s %s\n" % (data._tables, data.name)) 

177 return data 2 ctx1cg

178 language = language if language else lang.check_file_language(source) 2 ctx1cg

179 # print(f"DEBUG: Dataloader TestCaseFile getting datafile language={language}") 

180 datafile = None 2 ctx1cg

181 try: 2 ctx1cg

182 datafile = robotapi.TestCaseFile(parent, source, settings, language).populate() 2 ctx1cg

183 except DataError: 2 ctx1cg

184 # print(f"DEBUG: Dataloader TestCaseFile testdata DataError source={source}") 

185 pass # We try once more in case is a Resource 2 ctx1cg

186 if datafile: 2 ctx1cg

187 # print(f"DEBUG: Dataloader TestCaseFile return datafile={datafile}") 

188 return datafile 1 ctx1c

189 if source.endswith(("resource", "robot")): 2 ctx1cg

190 datafile = robotapi.ResourceFile(source, settings, language).populate() 2 ctx1cg

191 # print(f"DEBUG: Dataloader returning TestCaseFile datafile={datafile}, type={type(datafile)}") 

192 return datafile 2 ctx1cg

193 

194 

195class ExcludedDirectory(robotapi.TestDataDirectory): 2 ctx1ab

196 def __init__(self, parent, path, language=None): 2 ctx1ab

197 self._parent = parent 

198 self._path = path 

199 self.language = language 

200 robotapi.TestDataDirectory.__init__(self, parent, path, language=self.language) 

201 

202 def has_tests(self): 2 ctx1ab

203 return True