Coverage for backend/django/core/auxiliary/models/Flowsheet.py: 97%

1from django.db import models 

2from django.utils import timezone 

3 

4from CoreRoot import settings 

5from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

6from authentication.user.models import User 

7from authentication.user.AccessTable import AccessTable 

8import core.auxiliary.enums.ViewType as ViewType 

9 

10 

11class Flowsheet(models.Model): 

12 name = models.CharField(max_length=64, default="Flowsheet") 

13 buildVersion = models.CharField(max_length=32, default=None, null=True) 

14 buildDate = models.CharField(max_length=32, default=None, null=True) 

15 savedDate = models.DateTimeField(null=True) 

16 # objectCounter = models.IntegerField(default=0) 

17 rootGrouping = models.ForeignKey("flowsheetInternals_graphicData.Grouping", default=None, null=True, on_delete=models.CASCADE, related_name="flowsheets") 

18 owner = models.ForeignKey(User, on_delete=models.CASCADE, related_name="flowsheets", null=True) 

19 flowsheet_template_type = models.CharField(max_length=32, choices=FlowsheetTemplateType.choices , default=FlowsheetTemplateType.NotTemplate) 

20 created_at = models.DateTimeField(auto_now_add=True) 

21 

22 @classmethod 

23 def create(cls, **kwargs): 

24 from flowsheetInternals.graphicData.models.groupingModel import Grouping 

25 from PinchAnalysis.models.StreamDataProject import StreamDataProject 

26 

27 project_number = Flowsheet.objects.count() + 1 

28 saved_date = timezone.now() 

29 

30 field_values = { 

31 'name': kwargs.get('name', f'Project-{project_number}'), 

32 'buildVersion': kwargs.get('buildVersion', 'Not set'), 

33 'buildDate': kwargs.get('buildDate', 'No Build Date Set'), 

34 'savedDate': saved_date, 

35 'owner': kwargs.get('owner') 

36 } 

37 

38 instance = Flowsheet.objects.create(**field_values) 

39 

40 AccessTable.objects.create( 

41 user=instance.owner, 

42 flowsheet=instance, 

43 ) 

44 

45 # create the root group 

46 rootGroup = Grouping.create(instance, group=None, componentName="Flowsheet", visible=True, isRoot=True) 

47 

48 instance.rootGrouping = rootGroup 

49 instance.save() 

50 

51 StreamDataProject.create(rootGroup, flowsheet=instance) 

52 return instance 

53 

54 def set_saved_date(self): 

55 self.savedDate = timezone.now() 

56 self.save() 

57 

58 

59 @classmethod 

60 def share_flowsheet(cls, owner, flowsheet_id, user_email): 

61 """ 

62 Share a flowsheet with a user by their email address. 

63 """ 

64 flowsheet = cls.objects.get(id=flowsheet_id, owner=owner) 

65 user = User.objects.get(email=user_email) 

66 

67 if user.email == settings.PLATFORM_TEST_EMAIL and not owner.is_tester: 

68 raise PermissionError("Only testers can share flowsheets with the platform test account.") 

69 

70 AccessTable.objects.create(user=user, flowsheet=flowsheet) 

71 

72 @classmethod 

73 def get_flowsheets_by_view_type(cls, user, view_type): 

74 # filter to exclude templates 

75 base_query = cls.objects.filter(flowsheet_template_type=FlowsheetTemplateType.NotTemplate) 

76 

77 if view_type == ViewType.ALL: 

78 return base_query.filter(access_list__user=user) 

79 elif view_type == ViewType.SHARED: 

80 return base_query.filter(access_list__user=user).exclude(owner=user) 

81 elif view_type == ViewType.OWNED: 

82 return base_query.filter(owner=user) 

83 else: 

84 return base_query.filter(owner=user) 

85 

86 @classmethod 

87 def get_shared_users(cls, owner, flowsheet_id): 

88 flowsheet = cls.objects.get(id=flowsheet_id, owner=owner) 

89 return AccessTable.objects.filter(flowsheet=flowsheet).exclude(user=owner).values_list('user__email', flat=True) 

90 

91 @classmethod 

92 def remove_user(cls, owner, flowsheet_id, user_email): 

93 """ 

94 Remove a user from the access list of the flowsheet. 

95 """ 

96 flowsheet = cls.objects.get(id=flowsheet_id, owner=owner) 

97 user = User.objects.get(email=user_email) 

98 if user.id == owner.id: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 raise ValueError("Cannot remove the owner of the flowsheet.") 

100 

101 AccessTable.objects.get(user=user, flowsheet=flowsheet).delete()