Coverage for src / mesh / ojs / import_file.py: 0%

1from pathlib import Path 

2from typing import TYPE_CHECKING, Protocol 

3 

4from django.core.files import File 

5from lxml import etree 

6 

7from mesh.models.orm.review_models import ( 

8 ReviewAdditionalFile, 

9) 

10from mesh.models.orm.submission_models import ( 

11 SubmissionAdditionalFile, 

12 # SubmissionMainFile, 

13 SubmissionMainFile, 

14) 

15from mesh.ojs.lxml_utils import ( 

16 find_or_raise, 

17 findtext_or_raise, 

18 get_or_raise, 

19 str_to_datetime, 

20) 

21 

22if TYPE_CHECKING: 

23 from io import BufferedReader 

24 

25 from mesh.models import SubmissionVersion 

26 from mesh.models.orm.file_models import BaseFileWrapperModel 

27 from mesh.models.orm.review_models import ( 

28 Review, 

29 # ReviewAdditionalFile, 

30 ) 

31 

32 

33class CreateFileLogic(Protocol): 

34 def __call__(self, file: "BufferedReader", filename: str) -> "BaseFileWrapperModel": ... 

35 

36 

37def version_main_file_factory(version: "SubmissionVersion"): 

38 """Returns a callable that creates a SubmissionMainFile for the given version""" 

39 

40 def inner(file: "BufferedReader", filename: str): 

41 try: 

42 version.main_file 

43 raise ValueError("Main file already exists") 

44 except SubmissionMainFile.DoesNotExist: 

45 pass 

46 file_instance = SubmissionMainFile( 

47 attached_to=version, 

48 ) 

49 file_instance.file.save(name=filename, content=File(file), save=False) 

50 return file_instance 

51 

52 return inner 

53 

54 

55def version_additional_file_factory(version: "SubmissionVersion"): 

56 """Returns a callable that creates a SubmissionAdditionalFile for the given version""" 

57 

58 def inner(file: "BufferedReader", filename: str): 

59 file_instance = SubmissionAdditionalFile( 

60 attached_to=version, 

61 ) 

62 file_instance.file.save(name=filename, content=File(file), save=False) 

63 return file_instance 

64 

65 return inner 

66 

67 

68def review_file_factory(review: "Review"): 

69 """Returns a callable that creates a ReviewAdditionalFile for the given review""" 

70 

71 def inner(file: "BufferedReader", filename: str): 

72 review_file = ReviewAdditionalFile(attached_to=review) 

73 review_file.file.save(name=filename, content=File(file), save=False) 

74 return review_file 

75 

76 return inner 

77 

78 

79def import_ojs_file(element: "etree._Element", callback: CreateFileLogic, user, path: str): 

80 """Import a submission_file or workflow_file. 

81 

82 Keyword arguments: 

83 element -- the real part (default 0.0) 

84 imag -- the imaginary part (default 0.0) 

85 Pass the result of file factory as second argument""" 

86 

87 file_element = find_or_raise(element, "file") 

88 href_element = find_or_raise(file_element, "href") 

89 filename = findtext_or_raise(element, "name") 

90 file_path = Path(path).parent / get_or_raise(href_element, "src") 

91 

92 date_created = str_to_datetime(get_or_raise(element, "created_at")) 

93 date_updated = str_to_datetime(get_or_raise(element, "updated_at")) 

94 with file_path.open(mode="rb") as f: 

95 file_instance = callback(f, filename) 

96 file_instance.override_saved_date( 

97 date_created=date_created, 

98 date_last_modified=date_updated, 

99 created_by_user=user, 

100 last_modified_by_user=user, 

101 ) 

102 file_instance.save()