Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions core/__init__.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
from core.utils import *
from core.datetimes.shared import *
from core.custom_lookups import *
from core.utils import * # noqa: F401,F403
from core.datetimes.shared import * # noqa: F401,F403
from core.custom_lookups import * # noqa: F401,F403

default_app_config = 'core.apps.CoreConfig'
default_app_config = "core.apps.CoreConfig"

# For IDE support, filled at runtime
datetime = None
exit
66 changes: 44 additions & 22 deletions core/abs_calculation_rule.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,9 @@ def get_calculation_rule_name(cls):
def set_calculation_rule_name(cls, val):
type(cls)._calculation_rule_name = val

calculation_rule_name = property(get_calculation_rule_name, set_calculation_rule_name)
calculation_rule_name = property(
get_calculation_rule_name, set_calculation_rule_name
)

@classmethod
def get_description(cls):
Expand All @@ -52,7 +54,9 @@ def get_impacted_class_parameter(cls):
def set_impacted_class_parameter(cls, val):
type(cls)._impacted_class_parameter = val

impacted_class_parameter = property(get_impacted_class_parameter, set_impacted_class_parameter)
impacted_class_parameter = property(
get_impacted_class_parameter, set_impacted_class_parameter
)

@classmethod
def get_type(cls):
Expand Down Expand Up @@ -87,8 +91,11 @@ def set_from_to(cls, val):
@classmethod
def ready(cls):
now = datetime.datetime.now()
condition_is_valid = (now >= cls.date_valid_from and now <= cls.date_valid_to) \
if cls.date_valid_to else (now >= cls.date_valid_from and cls.date_valid_to is None)
condition_is_valid = (
(now >= cls.date_valid_from and now <= cls.date_valid_to)
if cls.date_valid_to
else (now >= cls.date_valid_from and cls.date_valid_to is None)
)
if not condition_is_valid:
cls.status = "inactive"

Expand All @@ -111,14 +118,18 @@ def calculate(cls, instance, *args, **kwargs):
def get_linked_class(cls, sender, class_name, **kwargs):
# calculation are loaded on the side, therefore contentType have to be loaded on execution
from django.contrib.contenttypes.models import ContentType

list_class = []
if class_name is not None:
model_class = ContentType.objects.filter(model__iexact=class_name).first()
if model_class:
model_class = model_class.model_class()
list_class = list_class + \
[f.remote_field.model.__name__ for f in model_class._meta.fields
if f.get_internal_type() == 'ForeignKey' and f.remote_field.model.__name__ != "User"]
list_class = list_class + [
f.remote_field.model.__name__
for f in model_class._meta.fields
if f.get_internal_type() == "ForeignKey"
and f.remote_field.model.__name__ != "User"
]
else:
list_class.append("Calculation")
return list_class
Expand All @@ -143,29 +154,33 @@ def get_rule_details(cls, sender, class_name, **kwargs):
@classmethod
def get_parameters(cls, sender, class_name, instance, **kwargs):
"""
class_name is the class name of the object where the calculation param need to be added
instance is where the link with a calculation need to be found,
like the CPB in case of PH insuree or Contract Details
return a list only with rule details that matches step 1 and 2
class_name is the class name of the object where the calculation param need to be added
instance is where the link with a calculation need to be found,
like the CPB in case of PH insuree or Contract Details
return a list only with rule details that matches step 1 and 2
"""
rule_details = cls.get_rule_details(sender=sender, class_name=class_name)
if rule_details:
if cls.check_calculation(instance=instance):
return rule_details["parameters"] if "parameters" in rule_details else []
return (
rule_details["parameters"] if "parameters" in rule_details else []
)

@classmethod
def run_calculation_rules(cls, sender, instance, user, context, **kwargs):
"""
this function will send a signal and the rules will
reply if they have object matching the classname in their list of object
this function will send a signal and the rules will
reply if they have object matching the classname in their list of object
"""
list_class = cls.get_linked_class(sender, instance.__class__.__name__)
# if the class have a calculation param, (like contribution or payment plan) add class name
if hasattr(instance, 'calculation'):
if hasattr(instance, "calculation"):
list_class.append(instance.__class__.__name__)
if list_class:
for class_name in list_class:
rule_details = cls.get_rule_details(class_name=class_name, sender=sender)
rule_details = cls.get_rule_details(
class_name=class_name, sender=sender
)
if rule_details or len(cls.impacted_class_parameter) == 0:
# add context to kwargs
kwargs["context"] = context
Expand All @@ -174,32 +189,39 @@ def run_calculation_rules(cls, sender, instance, user, context, **kwargs):

@classmethod
def calculate_if_active_for_object(cls, instance, **kwargs):
if cls.active_for_object(instance=instance, context=kwargs['context']):
if cls.active_for_object(instance=instance, context=kwargs["context"]):
return cls.calculate(instance, **kwargs)

@classmethod
def run_convert(cls, instance, convert_to, **kwargs):
"""
execute the conversion for the instance with the first
rule that provide the conversion (see get_convert_from_to)
execute the conversion for the instance with the first
rule that provide the conversion (see get_convert_from_to)
"""
convert_from = instance.__class__.__name__
if convert_from == "Contract":
convert_from = "ContractContributionPlanDetails"
list_possible_conversion = cls.get_convert_from_to()
for possible_conversion in list_possible_conversion:
if convert_from == possible_conversion['from'] and convert_to == possible_conversion['to']:
if (
convert_from == possible_conversion["from"]
and convert_to == possible_conversion["to"]
):
result = cls.convert(instance=instance, convert_to=convert_to, **kwargs)
return result

@classmethod
def get_convert_from_to(cls):
"""
get the possible conversion, return [calc UUID, from, to]
get the possible conversion, return [calc UUID, from, to]
"""
list_possible_conversion = []
for ft in cls.from_to:
convert_from_to = {'calc_uuid': cls.uuid, 'from': ft['from'], 'to': ft['to']}
convert_from_to = {
"calc_uuid": cls.uuid,
"from": ft["from"],
"to": ft["to"],
}
list_possible_conversion.append(convert_from_to)
return list_possible_conversion

Expand Down
127 changes: 92 additions & 35 deletions core/apps.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,14 @@
"longstrfdate": "%a %d %B %Y",
"iso_raw_date": "False",
"age_of_majority": "18",
"async_mutations": "True" if os.environ.get("ASYNC", os.environ.get("MODE", "PROD")) == "PROD" else "False",
"async_mutations": (
"True"
if os.environ.get("ASYNC", os.environ.get("MODE", "PROD")) == "PROD"
else "False"
),
"password_reset_template": "password_reset.txt",
"currency": "$",
"gql_query_claim_admins_perms" : [],
"gql_query_claim_admins_perms": [],
"gql_query_users_perms": ["121701"],
"gql_mutation_create_users_perms": ["121702"],
"gql_mutation_update_users_perms": ["121703"],
Expand All @@ -52,20 +56,20 @@
"fields_controls_eo": {},
"is_valid_health_facility_contract_required": False,
"secondary_calendar": None,
"locked_user_password_hash": 'locked',
"locked_user_password_hash": "locked",
"gql_query_enable_viewing_masked_data_perms": ["900101"],
"csrf_protect_login": True,
}


class CoreConfig(AppConfig):
default_auto_field = 'django.db.models.AutoField' # Django 3.1+
default_auto_field = "django.db.models.AutoField" # Django 3.1+
name = MODULE_NAME
username_code_length = 12
username_changeable = True
age_of_majority = 18
password_reset_template = "password_reset.txt"

gql_query_claim_admins_perms = []
gql_query_roles_perms = []
gql_mutation_create_roles_perms = []
Expand Down Expand Up @@ -106,22 +110,27 @@ class CoreConfig(AppConfig):
csrf_protect_login = None

def _import_module(self, cfg, k):
logger.info('import %s.%s' %
(cfg["%s_module" % k], cfg["%s_package" % k]))
logger.info("import %s.%s" % (cfg["%s_module" % k], cfg["%s_package" % k]))
return importlib.import_module(
cfg["%s_module" % k], package=cfg["%s_package" % k])
cfg["%s_module" % k], package=cfg["%s_package" % k]
)

def _configure_calendar(self, cfg):
this.shortstrfdate = cfg["shortstrfdate"]
this.longstrfdate = cfg["longstrfdate"]
this.iso_raw_date = False if cfg["iso_raw_date"] is None else cfg["iso_raw_date"].lower(
) == "true"
this.iso_raw_date = (
False
if cfg["iso_raw_date"] is None
else cfg["iso_raw_date"].lower() == "true"
)
try:
this.calendar = self._import_module(cfg, "calendar")
this.datetime = self._import_module(cfg, "datetime")
except Exception:
logger.error('Failed to configure calendar, using default!\n%s: %s' % (
sys.exc_info()[0].__name__, sys.exc_info()[1]))
logger.error(
"Failed to configure calendar, using default!\n%s: %s"
% (sys.exc_info()[0].__name__, sys.exc_info()[1])
)
this.calendar = self._import_module(DEFAULT_CFG, "calendar")
this.datetime = self._import_module(DEFAULT_CFG, "datetime")

Expand All @@ -138,60 +147,107 @@ def _configure_currency(self, cfg):
this.currency = str(cfg["currency"])

def _configure_auto_provisioning(self, cfg):
if bool(os.environ.get('NO_DATABASE', False)):
logger.info('env NO_DATABASE set to True: no user auto provisioning possible!')
if bool(os.environ.get("NO_DATABASE", False)):
logger.info(
"env NO_DATABASE set to True: no user auto provisioning possible!"
)
return
group = cfg["auto_provisioning_user_group"]
this.auto_provisioning_user_group = group
try:
from .models import Group

Group.objects.get(name=group)
except Group.DoesNotExist:
g = Group(name=group)
g.save()
from django.contrib.auth.models import Permission

p = Permission.objects.get(codename="view_user")
g.permissions.add(p)
g.save()
except Exception as e:
logger.warning('Failed set auto_provisioning_user_group ' + str(e))
logger.warning("Failed set auto_provisioning_user_group " + str(e))

def _configure_graphql(self, cfg):
this.async_mutations = True if cfg["async_mutations"] is None else cfg["async_mutations"].lower() == "true"
this.async_mutations = (
True
if cfg["async_mutations"] is None
else cfg["async_mutations"].lower() == "true"
)

def _configure_permissions(self, cfg):
CoreConfig.gql_query_claim_admins_perms = cfg["gql_query_claim_admins_perms"]
CoreConfig.gql_query_roles_perms = cfg["gql_query_roles_perms"]
CoreConfig.gql_mutation_create_roles_perms = cfg["gql_mutation_create_roles_perms"]
CoreConfig.gql_mutation_update_roles_perms = cfg["gql_mutation_update_roles_perms"]
CoreConfig.gql_mutation_replace_roles_perms = cfg["gql_mutation_replace_roles_perms"]
CoreConfig.gql_mutation_duplicate_roles_perms = cfg["gql_mutation_duplicate_roles_perms"]
CoreConfig.gql_mutation_delete_roles_perms = cfg["gql_mutation_delete_roles_perms"]
CoreConfig.gql_mutation_create_roles_perms = cfg[
"gql_mutation_create_roles_perms"
]
CoreConfig.gql_mutation_update_roles_perms = cfg[
"gql_mutation_update_roles_perms"
]
CoreConfig.gql_mutation_replace_roles_perms = cfg[
"gql_mutation_replace_roles_perms"
]
CoreConfig.gql_mutation_duplicate_roles_perms = cfg[
"gql_mutation_duplicate_roles_perms"
]
CoreConfig.gql_mutation_delete_roles_perms = cfg[
"gql_mutation_delete_roles_perms"
]
CoreConfig.gql_query_users_perms = cfg["gql_query_users_perms"]
CoreConfig.gql_mutation_create_users_perms = cfg["gql_mutation_create_users_perms"]
CoreConfig.gql_mutation_update_users_perms = cfg["gql_mutation_update_users_perms"]
CoreConfig.gql_mutation_delete_users_perms = cfg["gql_mutation_delete_users_perms"]
CoreConfig.gql_query_enrolment_officers_perms = cfg["gql_query_enrolment_officers_perms"]
CoreConfig.gql_mutation_create_enrolment_officers_perms = cfg["gql_mutation_create_enrolment_officers_perms"]
CoreConfig.gql_mutation_update_enrolment_officers_perms = cfg["gql_mutation_update_enrolment_officers_perms"]
CoreConfig.gql_mutation_delete_enrolment_officers_perms = cfg["gql_mutation_delete_enrolment_officers_perms"]
CoreConfig.gql_query_claim_administrator_perms = cfg["gql_query_claim_administrator_perms"]
CoreConfig.gql_mutation_create_claim_administrator_perms = cfg["gql_mutation_create_claim_administrator_perms"]
CoreConfig.gql_mutation_update_claim_administrator_perms = cfg["gql_mutation_update_claim_administrator_perms"]
CoreConfig.gql_mutation_delete_claim_administrator_perms = cfg["gql_mutation_delete_claim_administrator_perms"]
CoreConfig.gql_mutation_delete_claim_administrator_perms = cfg["gql_mutation_delete_claim_administrator_perms"]
CoreConfig.gql_query_enable_viewing_masked_data_perms = cfg["gql_query_enable_viewing_masked_data_perms"]
CoreConfig.gql_mutation_create_users_perms = cfg[
"gql_mutation_create_users_perms"
]
CoreConfig.gql_mutation_update_users_perms = cfg[
"gql_mutation_update_users_perms"
]
CoreConfig.gql_mutation_delete_users_perms = cfg[
"gql_mutation_delete_users_perms"
]
CoreConfig.gql_query_enrolment_officers_perms = cfg[
"gql_query_enrolment_officers_perms"
]
CoreConfig.gql_mutation_create_enrolment_officers_perms = cfg[
"gql_mutation_create_enrolment_officers_perms"
]
CoreConfig.gql_mutation_update_enrolment_officers_perms = cfg[
"gql_mutation_update_enrolment_officers_perms"
]
CoreConfig.gql_mutation_delete_enrolment_officers_perms = cfg[
"gql_mutation_delete_enrolment_officers_perms"
]
CoreConfig.gql_query_claim_administrator_perms = cfg[
"gql_query_claim_administrator_perms"
]
CoreConfig.gql_mutation_create_claim_administrator_perms = cfg[
"gql_mutation_create_claim_administrator_perms"
]
CoreConfig.gql_mutation_update_claim_administrator_perms = cfg[
"gql_mutation_update_claim_administrator_perms"
]
CoreConfig.gql_mutation_delete_claim_administrator_perms = cfg[
"gql_mutation_delete_claim_administrator_perms"
]
CoreConfig.gql_mutation_delete_claim_administrator_perms = cfg[
"gql_mutation_delete_claim_administrator_perms"
]
CoreConfig.gql_query_enable_viewing_masked_data_perms = cfg[
"gql_query_enable_viewing_masked_data_perms"
]
CoreConfig.csrf_protect_login = cfg["csrf_protect_login"]

CoreConfig.fields_controls_user = cfg["fields_controls_user"]
CoreConfig.fields_controls_eo = cfg["fields_controls_eo"]

def _configure_additional_settings(self, cfg):
CoreConfig.is_valid_health_facility_contract_required = cfg["is_valid_health_facility_contract_required"]
CoreConfig.is_valid_health_facility_contract_required = cfg[
"is_valid_health_facility_contract_required"
]
CoreConfig.secondary_calendar = cfg["secondary_calendar"]

def ready(self):
from .models import ModuleConfiguration

cfg = ModuleConfiguration.get_or_default(MODULE_NAME, DEFAULT_CFG)
self._configure_calendar(cfg)
self._configure_user_config(cfg)
Expand All @@ -207,5 +263,6 @@ def ready(self):

# The scheduler starts as soon as it gets a job, which could be before Django is ready, so we enable it here
from core import scheduler

if settings.SCHEDULER_AUTOSTART:
scheduler.start()
Loading
Loading