Coverage for src/robotide/controller/project.py: 82%
324 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-06 10:40 +0100
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-06 10:40 +0100
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.
16import os 1ab
17import shutil 1ab
18import tempfile 1ab
20from .basecontroller import WithNamespace, _BaseController 1ab
21from .dataloader import DataLoader 1ab
22from .robotdata import new_test_case_file, new_test_data_directory 1ab
23from ..context import LOG 1ab
24from ..controller.ctrlcommands import NullObserver, SaveFile 1ab
25from ..publish.messages import RideOpenSuite, RideNewProject, RideFileNameChanged 1ab
26from .. import spec 1ab
27from ..spec.xmlreaders import SpecInitializer 1ab
30class Project(_BaseController, WithNamespace): 1ab
32 def __init__(self, namespace=None, settings=None, library_manager=None, tasks=False, file_language=None): 1ab
33 from .filecontrollers import ResourceFileControllerFactory 1akpqrstumojnclvw
34 self._library_manager = self._construct_library_manager(library_manager, settings) 1akpqrstumojnclvw
35 if not self._library_manager.is_alive(): 1akpqrstumojnclvw
36 self._library_manager.start() 1akpqrstumojnclvw
37 self._name_space = namespace 1akpqrstumojnclvw
38 self._set_namespace(self._name_space) 1akpqrstumojnclvw
39 self.tasks = tasks 1akpqrstumojnclvw
40 self.internal_settings = settings 1akpqrstumojnclvw
41 self._loader = DataLoader(self._name_space, settings) 1akpqrstumojnclvw
42 self.controller = None 1akpqrstumojnclvw
43 self.name = None 1akpqrstumojnclvw
44 self.external_resources = [] 1akpqrstumojnclvw
45 self.file_language = file_language 1akpqrstumojnclvw
46 self._resource_file_controller_factory = ResourceFileControllerFactory(self._name_space, self) 1akpqrstumojnclvw
47 self._serializer = Serializer(settings, LOG) 1akpqrstumojnclvw
49 @staticmethod 1ab
50 def _construct_library_manager(library_manager, settings): 1ab
51 return library_manager or \ 1akpqrstumojnclvw
52 spec.LibraryManager(spec.DATABASE_FILE, SpecInitializer(settings.get('library xml directories', [])[:]))
54 def __del__(self): 1ab
55 if self._library_manager: 2a HbPbQbO n 0bIb
56 self.close() 2a Hbn Ib
58 def close(self): 1ab
59 self._library_manager.stop() 2a Hbn Ib
60 self._library_manager = None 2a Hbn Ib
62 def _set_namespace(self, namespace): 1ab
63 namespace.set_library_manager(self._library_manager) 1akpqrstumojnclvw
64 WithNamespace._set_namespace(self, namespace) 1akpqrstumojnclvw
66 @property 1ab
67 def display_name(self): 1ab
68 return 'Project' 1akl
70 @property 1ab
71 def default_dir(self): 1ab
72 return os.path.abspath(self.internal_settings.get('default directory', ''))
74 def update_default_dir(self, path): 1ab
75 default_dir = path if os.path.isdir(path) else os.path.dirname(path) 1ax
76 self.internal_settings.set('default directory', default_dir) 1ax
77 self._name_space.update_exec_dir_global_var(default_dir) 1ax
78 self._name_space.update_cur_dir_global_var(default_dir) 1ax
80 # DEBUG: in all other controllers data returns a robot data model object.
81 @property 1ab
82 def data(self): 1ab
83 return self.controller 2a 1bP Q z x A y JbKbLbO MbNbJ ObR S T U V W X Y Z 0 1 2 3 4 5 6 7 8 9 ! # $ % ' p q r s t u 2bL m o j n ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCbd e i f g h Db3bEb4b5b6b7b8bFb
85 @property 1ab
86 def suite(self): 1ab
87 return self.controller.data if self.controller else None 2a 9b!b#b$b
89 @property 1ab
90 def datafiles(self): 1ab
91 return self._suites() + self.resources 2a P Q z x A y J R S T U V W X Y Z 0 1 2 3 4 5 6 7 8 9 ! # $ % ' p q r s t u L m o j ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCbd e i f g h DbEbFb
93 @property 1ab
94 def resources(self): 1ab
95 return self._resource_file_controller_factory.resources 2a P Q z x A y JbKbLbGbO MbNbJ R S T U V W X Y Z 0 1 2 3 4 5 6 7 8 9 ! # $ % ' p q r s t u RbL m o j ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCbd e i f g h DbEb%bFb
97 @property 1ab
98 def resource_file_controller_factory(self): 1ab
99 return self._resource_file_controller_factory 1azxAyJpqrstumojncdeifgh
101 def find_controller_by_longname(self, longname, testname=None): 1ab
102 return self.controller.find_controller_by_longname(longname, testname) 2Pb'b(b)bQb
104 def get_language_from_settings(self): 1ab
105 from ..preferences import RideSettings 1x
106 _settings = RideSettings() 1x
107 lang = _settings.get('doc language', '') 1x
108 self.file_language = lang 1x
109 return lang 1x
111 def new_directory_project(self, path, tasks=False, lang=''): 1ab
112 lang = lang if lang else self.get_language_from_settings()
113 self._new_project(new_test_data_directory(path, tasks=tasks, lang=lang), tasks=tasks)
115 def new_file_project(self, path, tasks=False, lang=''): 1ab
116 lang = lang if lang else self.get_language_from_settings() 1x
117 self._new_project(new_test_case_file(path, tasks=tasks, lang=lang), tasks=tasks) 1x
119 def _new_project(self, datafile, tasks=False): 1ab
120 from .filecontrollers import data_controller, ResourceFileControllerFactory 1x
121 self.update_default_dir(datafile.directory) 1x
122 self.controller = data_controller(datafile, self, tasks=tasks) 1x
123 self._resource_file_controller_factory = ResourceFileControllerFactory(self.namespace, self) 1x
124 RideNewProject(path=datafile.source, datafile=datafile).publish() 1x
126 def new_resource(self, path, parent=None): 1ab
127 res = self.namespace.new_resource(path) 1zyjdeifgh
128 resource_controller = self._create_resource_controller(res, parent) 1zyjdeifgh
129 return resource_controller 1zyjdeifgh
131 def load_data(self, path, load_observer=None): 1ab
132 """ DEBUG: To be used in Localization
133 """
134 from robotide.context import APP 1aBkCDEFGHl
135 try: 1aBkCDEFGHl
136 robot_version = APP.robot_version 1aBkCDEFGHl
137 except AttributeError: 1aBkCDEFGHl
138 robot_version = '7.0.1' # It is failing at unit tests 1aBkCDEFGHl
139 print(f"DEBUG: project.py Project load_data robot version = {robot_version}") 1aBkCDEFGHl
140 from ..lib.compat.parsing.language import check_file_language 1aBkCDEFGHl
141 self.file_language = check_file_language(path) 1aBkCDEFGHl
142 load_observer = load_observer or NullObserver() 1aBkCDEFGHl
143 if self._load_initfile(path, load_observer, self.file_language): 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true1aBkCDEFGHl
144 return
145 if self._load_datafile(path, load_observer, self.file_language): 1aBkCDEFGHl
146 return
147 if self._load_resource(path, load_observer, self.file_language): 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 return
149 try:
150 load_observer.error("Given file '%s' is not a valid Robot Framework "
151 "test case or resource file." % path)
152 except AttributeError: # DEBUG
153 # print(f"DEBUG: load_data error not valid datafile: {path}")
154 pass
156 def is_excluded(self, source=None): 1ab
157 return self.internal_settings.excludes.contains(source) if self.internal_settings else False
159 def _load_initfile(self, path, load_observer, language=None): 1ab
160 if os.path.splitext(os.path.split(path)[1])[0] != '__init__': 160 ↛ 162line 160 didn't jump to line 162 because the condition on line 160 was always true1aBkCDEFGHl
161 return None 1aBkCDEFGHl
162 initfile = self._loader.load_initfile(path, load_observer, language)
163 if not initfile:
164 return None
165 self._populate_from_datafile(path, initfile, load_observer, language)
166 return initfile
168 def load_datafile(self, path, load_observer, language=None): 1ab
169 datafile = self._load_datafile(path, load_observer, language) 2a SbTbUbVbWbv w
170 if datafile:
171 return datafile
172 # print(f"DEBUG: project Before testing Resource load_datafile path={path}")
173 # Let's see if it is a .robot file valid as .resource
174 if path.endswith((".robot", ".resource")): 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 datafile = self._load_resource(path, load_observer, language)
176 if datafile:
177 return datafile
178 load_observer.error("Invalid data file '%s'." % path)
179 e = UserWarning("Invalid data file")
180 return e
182 def _load_datafile(self, path, load_observer, language=None): 1ab
183 # print(f"DEBUG: project ENTER _load_datafile path={path} self.file_language={self.file_language}"
184 # f" language={language}")
185 datafile = self._loader.load_datafile(path, load_observer, self.file_language) 2a B k C D SbE TbF G H UbVbWbl v w
186 if not datafile:
187 return None
188 self._populate_from_datafile(path, datafile, load_observer, language)
189 return datafile
191 def _populate_from_datafile(self, path, datafile, load_observer, language=None): 1ab
192 self.__init__(self.namespace, self.internal_settings, library_manager=self._library_manager,
193 file_language=language)
194 resources = self._loader.resources_for(datafile, load_observer)
195 self._create_controllers(datafile, resources)
196 RideOpenSuite(path=path, datafile=self.controller).publish()
197 load_observer.finish()
199 def _create_controllers(self, datafile, resources): 1ab
200 from .filecontrollers import data_controller
201 self.clear_namespace_update_listeners()
202 self.controller = data_controller(datafile, self)
203 new_resource_controllers = []
204 for r in resources:
205 self._create_resource_controller(r, resource_created_callback=lambda r_controller:
206 new_resource_controllers.append(r_controller))
207 for controller in new_resource_controllers:
208 self._inform_resource_created(controller)
210 def load_resource(self, path, load_observer, language=None): 1ab
211 resource = self._load_resource(path, load_observer, language) 2a XbM Yb
212 if resource: 1aM
213 return resource
214 load_observer.error("Invalid resource file '%s'." % path) 1M
216 def _load_resource(self, path, load_observer, language=None): 1ab
217 resource = self._loader.load_resource_file(path, load_observer, language) 2a XbM Yb
218 if not resource: 1aM
219 return None 1aM
220 ctrl = self._create_resource_controller(resource)
221 load_observer.finish()
222 return ctrl
224 def _create_resource_controller(self, parsed_resource, parent=None, 1ab
225 resource_created_callback=None):
226 old = self._resource_file_controller_factory.find(parsed_resource) 1azAyjdeifgh
227 if old: 1azAyjdeifgh
228 return old 1aydefgh
229 controller = self._resource_file_controller_factory.create(parsed_resource, parent=parent) 1azAyjdeifgh
230 self.insert_into_suite_structure(controller) 1azAyjdeifgh
231 resource_created_callback = resource_created_callback or self._inform_resource_created 1azAyjdeifgh
232 resource_created_callback(controller) 1azAyjdeifgh
233 return controller 1azAyjdeifgh
235 @staticmethod 1ab
236 def _inform_resource_created(controller): 1ab
237 controller.notify_opened() 1azAyjdeifgh
239 def insert_into_suite_structure(self, resource): 1ab
240 if self.controller and self.controller.is_inside_top_suite(resource): 2a z A y JbKbLbGbO MbNbj d e i f g h
241 self.controller.insert_to_test_data_directory(resource) 2a JbKbLbO MbNbd e i f g h
242 else:
243 self.external_resources.append(resource) 2a z A y Gbj
244 self._sort_external_resources() 2a z A y Gbj
246 def _sort_external_resources(self): 1ab
247 self.external_resources.sort(key=lambda resource: resource.name.lower()) 2a z A y Gbj
249 def get_all_keywords(self): 1ab
250 return self.get_all_keywords_from(ctrl.datafile for ctrl in self.datafiles if
251 ctrl.datafile)
253 def all_testcases(self): 1ab
254 for df in self._suites():
255 for test in df.tests:
256 yield test
258 def get_files_without_format(self, controller=None): 1ab
259 if controller:
260 controller_list = [controller]
261 else:
262 controller_list = self._get_all_dirty_controllers()
263 return [dc for dc in controller_list if dc.dirty and not dc.has_format()]
265 def get_root_suite_dir_path(self): 1ab
266 return self.suite.get_dir_path()
268 def is_dirty(self): 1ab
269 if self.data and self.is_datafile_dirty(self.data):
270 return True
271 for res in self.resources:
272 if res.dirty: 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true
273 return True
274 return False
276 def is_datafile_dirty(self, datafile): 1ab
277 if datafile.dirty: 2a J Ob
278 return True 2a Ob
279 for df in datafile.children: 279 ↛ 280line 279 didn't jump to line 280 because the loop on line 279 never started1aJ
280 if self.is_datafile_dirty(df):
281 return True
282 return False 1aJ
284 def change_format(self, controller, cformat): 1ab
285 if controller.is_same_format(cformat):
286 return
287 old_path = controller.filename
288 controller.set_format(cformat)
289 controller.execute(SaveFile(self.internal_settings.get('reformat', False)))
290 if old_path:
291 self._remove_file(old_path)
292 RideFileNameChanged(old_filename=old_path,
293 datafile=controller).publish()
295 def _remove_file(self, path): 1ab
296 if path and os.path.isfile(path):
297 os.remove(path)
299 def change_format_recursive(self, controller, cformat): 1ab
300 for datafile in controller.iter_datafiles():
301 self.change_format(datafile, cformat)
303 def remove_datafile(self, controller): 1ab
304 if controller is self.controller: 1mn
305 self.controller = None 1n
306 else:
307 self.controller.remove_child(controller) 1m
309 def remove_resource(self, controller): 1ab
310 self._resource_file_controller_factory.remove(controller) 2RbL j
312 def save(self, controller): 1ab
313 assert controller is not None 1cdeifgh
314 self._serializer.serialize_file(controller) 1cdeifgh
316 def _get_all_dirty_controllers(self): 1ab
317 return [controller for controller in self.datafiles if controller.dirty]
319 def _suites(self): 1ab
320 if not self.data: 2a P Q z x A y J R S T U V W X Y Z 0 1 2 3 4 5 6 7 8 9 ! # $ % ' p q r s t u L m o j ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCbd e i f g h DbEbFb
321 return [] 1azAymoj
322 return [df for df in self.data.iter_datafiles() if df not in self.resources] 2a P Q x J R S T U V W X Y Z 0 1 2 3 4 5 6 7 8 9 ! # $ % ' p q r s t u L o ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCbd e i f g h DbEbFb
324 def resource_import_modified(self, path, directory): 1ab
325 resource = self.namespace.get_resource(path, directory) 2A d e i f g h Zb
326 if resource: 2A d e i f g h Zb
327 return self._create_resource_controller(resource) 1Adefgh
329 def is_project_changed_from_disk(self): 1ab
330 from .filecontrollers import TestDataDirectoryController
331 for data_file in self.datafiles:
332 if isinstance(data_file, TestDataDirectoryController):
333 # print(f"DEBUG: Project is_project_changed_from_disk directory is {data_file.directory}")
334 if not os.path.exists(data_file.directory):
335 # print(f"DEBUG: Project is_project_changed_from_disk directory TRUE {data_file.directory}")
336 return True
337 else:
338 if data_file.has_been_modified_on_disk() or \
339 data_file.has_been_removed_from_disk():
340 # print(f"DEBUG: Project is_project_changed_from_disk FILE TRUE {data_file.name}")
341 return True
342 return False
345class Serializer(object): 1ab
347 def __init__(self, settings, logger): 1ab
348 self._settings = settings 1akpqrstumojnclvw
349 self._logger = logger 1akpqrstumojnclvw
350 self._errors = [] 1akpqrstumojnclvw
352 def serialize_file(self, controller): 1ab
353 try: 1cdeifgh
354 self._serialize_file(controller) 1cdeifgh
355 finally:
356 self._log_errors() 1cdeifgh
358 def _serialize_file(self, controller): 1ab
359 with Backup(controller): 1cdeifgh
360 try: 1cdeifgh
361 controller.datafile.save(**self._get_options()) 1cdeifgh
362 except Exception as err:
363 self._cache_error(controller, err)
364 raise
366 def _get_options(self): 1ab
367 return {'line_separator': self._get_line_separator(), 1cdeifgh
368 'pipe_separated': self._get_pipe_separated(),
369 'txt_separating_spaces': self._get_separating_spaces()}
371 def _get_line_separator(self): 1ab
372 setting = self._settings.get('line separator', 'native').lower() 1cdeifgh
373 if setting in ('crlf', 'windows'): 373 ↛ 374line 373 didn't jump to line 374 because the condition on line 373 was never true1cdeifgh
374 return '\r\n'
375 if setting in ('lf', 'unix'): 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true1cdeifgh
376 return '\n'
377 return os.linesep 1cdeifgh
379 def _get_pipe_separated(self): 1ab
380 return self._settings.get('txt format separator', 'space') == 'pipe' 1cdeifgh
382 def _get_separating_spaces(self): 1ab
383 return self._settings.get('txt number of spaces', 4) 1cdeifgh
385 def _cache_error(self, data, error): 1ab
386 self._errors.append("Error in serializing '%s':\n%s"
387 % (data.data.source, str(error)))
389 def _log_errors(self): 1ab
390 if self._errors: 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true1cdeifgh
391 self._logger.error('\n\n'.join(self._errors))
392 self._errors = []
395class Backup(object): 1ab
397 def __init__(self, file_controller): 1ab
398 self._path = file_controller.filename 1acdeifgh
399 self._file_controller = file_controller 1acdeifgh
400 self._backup = self._get_backup_name(self._path) 1acdeifgh
402 def _get_backup_name(self, path): 1ab
403 if path is None or not os.path.isfile(path): 1acdeifgh
404 return None 1adeifgh
405 self._backup_dir = tempfile.mkdtemp(dir=os.path.dirname(path)) 1c
406 return os.path.join(self._backup_dir, os.path.basename(path)) 1c
408 def __enter__(self): 1ab
409 self._make_backup() 1NIKcdeifgh
411 def _make_backup(self): 1ab
412 if self._backup: 1NIKcdeifgh
413 try: 1NIKc
414 self._move(self._path, self._backup) 1NIKc
415 except IOError: 1K
416 self._backup = None 1K
417 self._remove_backup_dir() 1K
419 def __exit__(self, *args): 1ab
420 if any(args): 1NIKcdeifgh
421 self._restore_backup() 1I
422 else:
423 self._remove_backup() 1NKcdeifgh
425 def _remove_backup(self): 1ab
426 if self._backup: 1cdeifgh
427 os.remove(self._backup) 1c
428 self._remove_backup_dir() 1c
430 def _restore_backup(self): 1ab
431 if self._backup: 431 ↛ exitline 431 didn't return from function '_restore_backup' because the condition on line 431 was always true1I
432 self._move(self._backup, self._path) 1I
433 self._file_controller.refresh_stat() 1I
434 self._remove_backup_dir() 1I
436 def _move(self, from_path, to_path): 1ab
437 shutil.move(from_path, to_path) 1c
439 def _remove_backup_dir(self): 1ab
440 shutil.rmtree(self._backup_dir) 1c