fix: remove frozen flag definitions and centralise CRE_ALLOW_IMPORT gate

application/config.py

@@ -1,18 +1,13 @@
 import os
 
 basedir = os.path.abspath(os.path.dirname(__file__))
-ENABLE_MYOPENCRE = os.getenv("ENABLE_MYOPENCRE", "false").lower() == "true"
 
 
 class Config:
     SQLALCHEMY_TRACK_MODIFICATIONS = False
     SQLALCHEMY_RECORD_QUERIES = False
     ITEMS_PER_PAGE = 20
     SLOW_DB_QUERY_TIME = 0.5
-    # Feature toggle for gap analysis optimization (default: False for safety)
-    GAP_ANALYSIS_OPTIMIZED = (
-        os.environ.get("GAP_ANALYSIS_OPTIMIZED", "False").lower() == "true"
-    )
 
 
 class DevelopmentConfig(Config):

application/feature_flags.py

@@ -0,0 +1,7 @@
+import os
+
+TRUE_VALUES = {"1", "true", "yes"}
+
+
+def is_cre_import_allowed() -> bool:
+    return os.getenv("CRE_ALLOW_IMPORT", "").strip().lower() in TRUE_VALUES

application/tests/gap_analysis_db_test.py

@@ -35,11 +35,8 @@ def cypher_side_effect(query, params=None, resolve_objects=True):
 
         self.mock_cypher.side_effect = cypher_side_effect
 
-        # Call the function with tiered pruning enabled
-        with patch(
-            "application.config.Config.GAP_ANALYSIS_OPTIMIZED", True, create=True
-        ):
-            db.NEO_DB.gap_analysis("StandardA", "StandardB")
+        # Call the function directly; tiered pruning is always-on
+        db.NEO_DB.gap_analysis("StandardA", "StandardB")
 
         # ASSERTION:
         # We expect cypher_query to be called.

application/tests/web_main_test.py

@@ -974,8 +974,6 @@ def test_all_cres(self, db_mock) -> None:
             )
 
     def test_import_from_cre_csv(self) -> None:
-        os.environ["CRE_ALLOW_IMPORT"] = "True"
-
         input_data, _ = data_gen.export_format_data()
         workspace = tempfile.mkdtemp()
         data = {}
@@ -987,19 +985,20 @@ def test_import_from_cre_csv(self) -> None:
 
         data["cre_csv"] = open(os.path.join(workspace, "cre.csv"), "rb")
 
-        with self.app.test_client() as client:
-            response = client.post(
-                "/rest/v1/cre_csv_import",
-                data=data,
-                buffered=True,
-                content_type="multipart/form-data",
-            )
-            print(f"\nSTATUS CODE: {response.status_code}, DATA: {response.data}")
-            self.assertEqual(200, response.status_code)
-            data = json.loads(response.data)
-            self.assertEqual("success", data.get("status"))
-            self.assertGreaterEqual(data.get("new_standards"), 2)
-            self.assertIsInstance(data.get("new_cres"), list)
+        with patch.dict(os.environ, {"CRE_ALLOW_IMPORT": "True"}):
+            with self.app.test_client() as client:
+                response = client.post(
+                    "/rest/v1/cre_csv_import",
+                    data=data,
+                    buffered=True,
+                    content_type="multipart/form-data",
+                )
+                print(f"\nSTATUS CODE: {response.status_code}, DATA: {response.data}")
+                self.assertEqual(200, response.status_code)
+                data = json.loads(response.data)
+                self.assertEqual("success", data.get("status"))
+                self.assertGreaterEqual(data.get("new_standards"), 2)
+                self.assertIsInstance(data.get("new_cres"), list)
 
     def test_get_cre_csv(self) -> None:
         # empty string means temporary db

application/web/web_main.py

@@ -22,6 +22,7 @@
 from application.cmd import cre_main
 from application.defs import cre_defs as defs
 from application.defs import cre_exceptions
+from application.feature_flags import is_cre_import_allowed
 
 from application.utils import spreadsheet as sheet_utils
 from application.utils import mdutils, redirectors, gap_analysis
@@ -722,11 +723,7 @@ def login_r(*args, **kwargs):
 def admin_imports_enabled_required(f):
     @wraps(f)
     def enabled_r(*args, **kwargs):
-        if str(os.environ.get("CRE_ALLOW_IMPORT", "")).lower() not in (
-            "1",
-            "true",
-            "yes",
-        ):
+        if not is_cre_import_allowed():
             abort(404, description="Admin imports API is disabled")
         return f(*args, **kwargs)
 
@@ -1119,7 +1116,7 @@ def get_cre_csv() -> Any:
 
 @app.route("/rest/v1/config", methods=["GET"])
 def get_config() -> Any:
-    return jsonify({"CRE_ALLOW_IMPORT": os.environ.get("CRE_ALLOW_IMPORT") == "1"})
+    return jsonify({"CRE_ALLOW_IMPORT": is_cre_import_allowed()})
 
 
 @app.route("/admin/imports/rerun", methods=["POST"])
@@ -1151,7 +1148,7 @@ def admin_imports_rerun() -> Any:
 
 @app.route("/rest/v1/cre_csv_import", methods=["POST"])
 def import_from_cre_csv() -> Any:
-    if str(os.environ.get("CRE_ALLOW_IMPORT", "")).lower() not in ("1", "true", "yes"):
+    if not is_cre_import_allowed():
         abort(
             403,
             "Importing is disabled, set the environment variable CRE_ALLOW_IMPORT to allow this functionality",