Coverage for .nox/test-3-9/lib/python3.9/site-packages/nskit/mixer/components/file.py: 93%

58 statements  

« prev     ^ index     » next       coverage.py v7.4.2, created at 2024-02-25 17:38 +0000

1"""File component.""" 

2from pathlib import Path 

3from typing import Any, Callable, Dict, Optional, Union 

4 

5from pydantic import Field 

6 

7from nskit.mixer.components.filesystem_object import FileSystemObject 

8from nskit.mixer.utilities import JINJA_ENVIRONMENT_FACTORY, Resource 

9 

10 

11class File(FileSystemObject): 

12 """File component.""" 

13 

14 content: Union[Resource, str, bytes, Path, Callable] = Field('', description='The file content') 

15 

16 def render_content(self, context: Dict[str, Any]): # pylint: disable=arguments-differ 

17 """Return the rendered content using the context and the Jinja environment.""" 

18 if context is None: 

19 context = {} 

20 if isinstance(self.content, Resource): 

21 content = self.content.load() 

22 elif isinstance(self.content, Path): 

23 with open(self.content) as f: 

24 content = f.read() 

25 elif isinstance(self.content, Callable): 

26 content = self.content(context) 

27 else: 

28 content = self.content 

29 if isinstance(content, str): 

30 # If it is a string, we render the content 

31 content = JINJA_ENVIRONMENT_FACTORY.environment.from_string(content).render(**context) 

32 return content 

33 

34 def write(self, base_path: Path, context: Dict[str, Any], override_path: Optional[Path] = None): 

35 """Write the rendered content to the appropriate path within the ``base_path``.""" 

36 file_path = self.get_path(base_path, context, override_path) 

37 content = self.render_content(context) 

38 response = {} 

39 if content is not None: 

40 if isinstance(content, str): 

41 open_str = 'w' 

42 elif isinstance(content, bytes): 

43 open_str = 'wb' 

44 with file_path.open(open_str) as output_file: 

45 output_file.write(content) 

46 response[file_path] = content 

47 return response 

48 

49 def dryrun(self, base_path: Path, context: Dict[str, Any], override_path: Optional[Path] = None): 

50 """Preview the file contents using the context.""" 

51 file_path = self.get_path(base_path, context, override_path) 

52 content = self.render_content(context) 

53 result = {} 

54 if content is not None: 

55 result[file_path] = content 

56 return result 

57 

58 def validate(self, base_path: Path, context: Dict[str, Any], override_path: Optional[Path] = None): 

59 """Validate the output against expected.""" 

60 missing = [] 

61 errors = [] 

62 ok = [] 

63 path = self.get_path(base_path, context, override_path) 

64 content = self.render_content(context) 

65 if content is not None: 

66 if not path.exists(): 

67 missing.append(path) 

68 else: 

69 if isinstance(content, bytes): 

70 read_str = 'rb' 

71 else: 

72 read_str = 'r' 

73 with open(path, read_str) as f: 

74 if f.read() != self.render_content(context): 

75 errors.append(path) 

76 else: 

77 ok.append(path) 

78 return missing, errors, ok