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 binned_at = models.DateTimeField(null=True) 

22 is_starred = models.BooleanField(default=False); 

23 is_binned = models.BooleanField(default=False); 

24 

25 @classmethod 

26 def create(cls, **kwargs): 

27 from flowsheetInternals.graphicData.models.groupingModel import Grouping 

28 from PinchAnalysis.models.StreamDataProject import StreamDataProject 

29 

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

31 saved_date = timezone.now() 

32 

33 field_values = { 

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

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

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

37 'savedDate': saved_date, 

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

39 } 

40 

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

42 

43 AccessTable.objects.create( 

44 user=instance.owner, 

45 flowsheet=instance, 

46 ) 

47 

48 # create the root group 

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

50 

51 instance.rootGrouping = rootGroup 

52 instance.save() 

53 

54 StreamDataProject.create(rootGroup, flowsheet=instance) 

55 return instance 

56 

57 def set_saved_date(self): 

58 self.savedDate = timezone.now() 

59 self.save() 

60 

61 

62 @classmethod 

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

64 """ 

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

66 """ 

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

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

69 

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

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

72 

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

74 

75 @classmethod 

76 def get_flowsheets_by_view_type(cls, user, view_type): 

77 # filter to exclude templates 

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

79 

80 if view_type == ViewType.ALL: 

81 return base_query.filter(access_list__user=user) 

82 elif view_type == ViewType.SHARED: 

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

84 elif view_type == ViewType.OWNED: 

85 return base_query.filter(owner=user) 

86 else: 

87 return base_query.filter(owner=user) 

88 

89 @classmethod 

90 def get_shared_users(cls, owner, flowsheet_id): 

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

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

93 

94 @classmethod 

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

96 """ 

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

98 """ 

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

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

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

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

103 

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