#637

Committed 21 May 2026 09:11AM UTC coverage: 88.055% (-0.02%) from 88.073%

Build # #637

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Feature/222 mobile rename saved to draft and improve statistics (#223)

* [#220] fix(sqlite): Phase 0+1 — fix defineSyncFormSubmissionTask and eliminate close/delete races

Phase 0 — Functional bug:
- background-task.js: defineSyncFormSubmissionTask now opens its own DB connection
  and passes it to syncFormSubmission (was called with undefined, always returned Failed)

Phase 1 — Crash eliminators:
- background-task.js: remove syncFormVersion internal db.closeAsync (ownership violation,
  caused Sentry #7286813050 — NativeStatement.finalizeAsync on closed resource)
- background-task.js: add finally-close to defineSyncFormVersionTask, registerBackgroundTask,
  syncDatapointsBackground (was scattered across early returns and success path only)
- cascades.js: fix dropFiles forEach(async) → reduce (fire-and-forget deletions)
- cascades.js: fix close/delete race — openDatabaseSync → closeSync before deleteAsync
- cascades.js: fix loadDataSource leaked connection — add outer try/finally closeAsync

Co-Authored-By: Claude <noreply@anthropic.com>

* [#220] fix(sqlite): Phase 2 — replace forEach(async) with awaited reduce

- FormPage.js: refreshForm made async; cascade close converted from
  forEach+closeAsync to reduce+closeSync; all callers now await refreshForm()
- AddUser.js: handleGetAllForms sequential form+cascade fetching via reduce
  with inner Promise.allSettled for concurrent cascade downloads
- AuthForm.js: handleGetAllForms parallel fetch via Promise.allSettled,
  sequential DB write via reduce to prevent overlapping upserts

Co-Authored-By: Claude <noreply@anthropic.com>

* [#220] fix(sqlite): Phase 3 — await fire-and-forget DB write, catch unhandled rejection

- SettingsForm.js: handleOnSwitch made async; DB write awaited inside try/catch
  with Sentry capture to prevent unhandled rejection crash on switch toggle
- SyncService.js: setInterval onSync() now has .catch(Sentry.captureException)
  to prevent unhandled rejection from crashing the app ... (continued)

Coverage Stats

4997 of 5827 branches covered (85.76%)

Branch coverage included in aggregate %.

9629 of 10783 relevant lines covered (89.3%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.58

backend/utils/custom_generator.py

import os
import sqlite3
import time
import pandas as pd
import logging
from django.conf import settings
from mis.settings import MASTER_DATA, STORAGE_PATH, COUNTRY_NAME
from api.v1.v1_profile.models import Administration

logger = logging.getLogger(__name__)


def generate_sqlite(model, test: bool = False):
    if not test:
        test = settings.TEST_ENV
    table_name = model._meta.db_table
    field_names = [f.name for f in model._meta.fields]
    objects = model.objects.all()
    file_name = "{0}/{1}{2}.sqlite".format(
        MASTER_DATA,
        "test_" if test else "",
        table_name,
    )
    if os.path.exists(file_name):
        try:
            os.remove(file_name)
        except OSError:
            pass
    data = pd.DataFrame(list(objects.values(*field_names)))
    no_rows = data.shape[0]
    if no_rows < 1:
        return
    # Add full_path_name for Administration model
    if model.__name__ == "Administration":
        # Get all Administration objects with their full_path_name property
        full_path_names = {
            obj.id: obj.full_path_name.replace("|", " - ") for obj in objects
        }
        # Add the full_path_name column to the DataFrame
        data["full_path_name"] = data["id"].apply(
            lambda id_: full_path_names.get(id_, "")
        )

    if "parent" in field_names:
        data["parent"] = data["parent"].apply(
            lambda x: int(x) if x == x else 0
        )
    elif "administration" in field_names:
        data["parent"] = data["administration"].apply(
            lambda x: int(x) if x == x else 0
        )
    else:
        data["parent"] = 0
    last_err = None
    for attempt in range(5):
        try:
            conn = sqlite3.connect(file_name, timeout=30)
            try:
                data.to_sql("nodes", conn, if_exists="replace", index=False)
            finally:
                conn.close()
            return file_name
        except sqlite3.OperationalError as err:
            last_err = err
            time.sleep(0.1 * (2 ** attempt))
    raise last_err


def update_sqlite(model, data, id=None):
    test = settings.TEST_ENV
    table_name = model._meta.db_table
    fields = data.keys()
    field_names = ", ".join([f for f in fields])
    placeholders = ", ".join(["?" for _ in range(len(fields))])
    update_placeholders = ", ".join([f"{f} = ?" for f in fields])
    params = list(data.values())
    if id:
        params += [id]
    file_name = "{0}/{1}{2}.sqlite".format(
        MASTER_DATA,
        "test_" if test else "",
        table_name,
    )
    conn = sqlite3.connect(file_name, timeout=30)
    try:
        with conn:
            c = conn.cursor()
            if id:
                c.execute("SELECT * FROM nodes WHERE id = ?", (id,))
                if c.fetchone():
                    query = f"UPDATE nodes \
                        SET {update_placeholders} WHERE id = ?"
                    c.execute(query, params)
            if not id:
                query = f"INSERT INTO nodes({field_names}) \
                    VALUES ({placeholders})"
                c.execute(query, params)
    except sqlite3.OperationalError:
        generate_sqlite(model=model, test=test)
    finally:
        conn.close()


def administration_csv_add(data: dict):
    test = settings.TEST_ENV
    filename = "{0}-administration.csv".format(
        "test" if test else COUNTRY_NAME
    )
    filepath = f"{STORAGE_PATH}/master_data/{filename}"
    if os.path.exists(filepath):
        df = pd.read_csv(filepath)
        new_data = {}
        if data.path:
            parent_ids = list(filter(lambda path: path, data.path.split(".")))
            parents = Administration.objects.filter(
                pk__in=parent_ids, level__id__gt=1
            ).all()
            for p in parents:
                new_data[p.level.name.lower()] = p.name
                new_data[f"{p.level.name.lower()}_id"] = p.id
        new_data[data.level.name.lower()] = data.name
        new_data[f"{data.level.name.lower()}_id"] = data.id
        new_df = pd.DataFrame([new_data])
        df = pd.concat([df, new_df], ignore_index=True)
        df.to_csv(filepath, index=False)
        return filepath
    else:
        logger.error(
            {
                "context": "insert_administration_row_csv",
                "message": (
                    f"{('test' if test else COUNTRY_NAME)}-administration.csv"
                    " doesn't exist"
                ),
            }
        )  # pragma: no cover
    return None


def find_index_by_id(df, id):
    for idx, row in df.iterrows():
        last_non_null_col = row.last_valid_index()
        last_non_null_value = row[last_non_null_col]
        if last_non_null_value == id:
            return idx
    return None


def administration_csv_update(data: dict):
    test = settings.TEST_ENV
    filename = "{0}-administration.csv".format(
        "test" if test else COUNTRY_NAME
    )
    filepath = f"{STORAGE_PATH}/master_data/{filename}"
    if os.path.exists(filepath):
        df = pd.read_csv(filepath)
        index = find_index_by_id(df=df, id=data.pk)
        if index is not None:
            if data.path:
                parent_ids = list(
                    filter(lambda path: path, data.path.split("."))
                )
                parents = Administration.objects.filter(
                    pk__in=parent_ids, level__id__gt=1
                ).all()
                for p in parents:
                    df.loc[index, p.level.name.lower()] = p.name
                    df.loc[index, f"{p.level.name.lower()}_id"] = p.id
            df.loc[index, data.level.name.lower()] = data.name
            df.loc[index, f"{data.level.name.lower()}_id"] = data.id
        df.to_csv(filepath, index=False)
        return filepath
    else:
        logger.error(
            {
                "context": "update_administration_row_csv",
                "message": f"{filename} doesn't exist",
            }
        )  # pragma: no cover
    return None


def administration_csv_delete(id: int):
    test = settings.TEST_ENV
    filename = "{0}-administration.csv".format(
        "test" if test else COUNTRY_NAME
    )
    filepath = f"{STORAGE_PATH}/master_data/{filename}"
    if os.path.exists(filepath):
        df = pd.read_csv(filepath)
        ix = find_index_by_id(df=df, id=id)
        if ix is not None:
            df.drop(index=ix, inplace=True)
        df.to_csv(filepath, index=False)
        return filepath
    else:
        logger.error(
            {
                "context": "delete_administration_row_csv",
                "message": f"{filename} doesn't exist",
            }
        )  # pragma: no cover
    return None

1	import os	1✔
2	import sqlite3	1✔
3	import time	1✔
4	import pandas as pd	1✔
5	import logging	1✔
6	from django.conf import settings	1✔
7	from mis.settings import MASTER_DATA, STORAGE_PATH, COUNTRY_NAME	1✔
8	from api.v1.v1_profile.models import Administration	1✔
9
10	logger = logging.getLogger(__name__)	1✔
11
12
13	def generate_sqlite(model, test: bool = False):	1✔
14	if not test:	1✔
15	test = settings.TEST_ENV	1✔
16	table_name = model._meta.db_table	1✔
17	field_names = [f.name for f in model._meta.fields]	1✔
18	objects = model.objects.all()	1✔
19	file_name = "{0}/{1}{2}.sqlite".format(	1✔
20	MASTER_DATA,
21	"test_" if test else "",
22	table_name,
23	)
24	if os.path.exists(file_name):	1✔
25	try:	1✔
26	os.remove(file_name)	1✔
27	except OSError:	×
28	pass	×
29	data = pd.DataFrame(list(objects.values(*field_names)))	1✔
30	no_rows = data.shape[0]	1✔
31	if no_rows < 1:	1✔
32	return	1✔
33	# Add full_path_name for Administration model
34	if model.__name__ == "Administration":	1✔
35	# Get all Administration objects with their full_path_name property
36	full_path_names = {	1✔
37	obj.id: obj.full_path_name.replace("\|", " - ") for obj in objects
38	}
39	# Add the full_path_name column to the DataFrame
40	data["full_path_name"] = data["id"].apply(	1✔
41	lambda id_: full_path_names.get(id_, "")
42	)
43
44	if "parent" in field_names:	1✔
45	data["parent"] = data["parent"].apply(	1✔
46	lambda x: int(x) if x == x else 0
47	)
48	elif "administration" in field_names:	1✔
49	data["parent"] = data["administration"].apply(	1✔
50	lambda x: int(x) if x == x else 0
51	)
52	else:
53	data["parent"] = 0	1✔
54	last_err = None	1✔
55	for attempt in range(5):	1!
56	try:	1✔
57	conn = sqlite3.connect(file_name, timeout=30)	1✔
58	try:	1✔
59	data.to_sql("nodes", conn, if_exists="replace", index=False)	1✔
60	finally:
61	conn.close()	1✔
62	return file_name	1✔
63	except sqlite3.OperationalError as err:	×
64	last_err = err	×
65	time.sleep(0.1 * (2 ** attempt))	×
66	raise last_err	×
67
68
69	def update_sqlite(model, data, id=None):	1✔
70	test = settings.TEST_ENV	1✔
71	table_name = model._meta.db_table	1✔
72	fields = data.keys()	1✔
73	field_names = ", ".join([f for f in fields])	1✔
74	placeholders = ", ".join(["?" for _ in range(len(fields))])	1✔
75	update_placeholders = ", ".join([f"{f} = ?" for f in fields])	1✔
76	params = list(data.values())	1✔
77	if id:	1✔
78	params += [id]	1✔
79	file_name = "{0}/{1}{2}.sqlite".format(	1✔
80	MASTER_DATA,
81	"test_" if test else "",
82	table_name,
83	)
84	conn = sqlite3.connect(file_name, timeout=30)	1✔
85	try:	1✔
86	with conn:	1✔
87	c = conn.cursor()	1✔
88	if id:	1✔
89	c.execute("SELECT * FROM nodes WHERE id = ?", (id,))	1✔
90	if c.fetchone():	1✔
91	query = f"UPDATE nodes \	1✔
92	SET {update_placeholders} WHERE id = ?"
93	c.execute(query, params)	1✔
94	if not id:	1✔
95	query = f"INSERT INTO nodes({field_names}) \	1✔
96	VALUES ({placeholders})"
97	c.execute(query, params)	1✔
98	except sqlite3.OperationalError:	1✔
99	generate_sqlite(model=model, test=test)	1✔
100	finally:
101	conn.close()	1✔
102
103
104	def administration_csv_add(data: dict):	1✔
105	test = settings.TEST_ENV	1✔
106	filename = "{0}-administration.csv".format(	1✔
107	"test" if test else COUNTRY_NAME
108	)
109	filepath = f"{STORAGE_PATH}/master_data/{filename}"	1✔
110	if os.path.exists(filepath):	1✔
111	df = pd.read_csv(filepath)	1✔
112	new_data = {}	1✔
113	if data.path:	1!
114	parent_ids = list(filter(lambda path: path, data.path.split(".")))	1✔
115	parents = Administration.objects.filter(	1✔
116	pk__in=parent_ids, level__id__gt=1
117	).all()
118	for p in parents:	1✔
119	new_data[p.level.name.lower()] = p.name	1✔
120	new_data[f"{p.level.name.lower()}_id"] = p.id	1✔
121	new_data[data.level.name.lower()] = data.name	1✔
122	new_data[f"{data.level.name.lower()}_id"] = data.id	1✔
123	new_df = pd.DataFrame([new_data])	1✔
124	df = pd.concat([df, new_df], ignore_index=True)	1✔
125	df.to_csv(filepath, index=False)	1✔
126	return filepath	1✔
127	else:
128	logger.error(
129	{
130	"context": "insert_administration_row_csv",
131	"message": (
132	f"{('test' if test else COUNTRY_NAME)}-administration.csv"
133	" doesn't exist"
134	),
135	}
136	) # pragma: no cover
137	return None	×
138
139
140	def find_index_by_id(df, id):	1✔
141	for idx, row in df.iterrows():	1✔
142	last_non_null_col = row.last_valid_index()	1✔
143	last_non_null_value = row[last_non_null_col]	1✔
144	if last_non_null_value == id:	1✔
145	return idx	1✔
146	return None	1✔
147
148
149	def administration_csv_update(data: dict):	1✔
150	test = settings.TEST_ENV	1✔
151	filename = "{0}-administration.csv".format(	1✔
152	"test" if test else COUNTRY_NAME
153	)
154	filepath = f"{STORAGE_PATH}/master_data/{filename}"	1✔
155	if os.path.exists(filepath):	1✔
156	df = pd.read_csv(filepath)	1✔
157	index = find_index_by_id(df=df, id=data.pk)	1✔
158	if index is not None:	1✔
159	if data.path:	1!
160	parent_ids = list(	1✔
161	filter(lambda path: path, data.path.split("."))
162	)
163	parents = Administration.objects.filter(	1✔
164	pk__in=parent_ids, level__id__gt=1
165	).all()
166	for p in parents:	1✔
167	df.loc[index, p.level.name.lower()] = p.name	1✔
168	df.loc[index, f"{p.level.name.lower()}_id"] = p.id	1✔
169	df.loc[index, data.level.name.lower()] = data.name	1✔
170	df.loc[index, f"{data.level.name.lower()}_id"] = data.id	1✔
171	df.to_csv(filepath, index=False)	1✔
172	return filepath	1✔
173	else:
174	logger.error(
175	{
176	"context": "update_administration_row_csv",
177	"message": f"{filename} doesn't exist",
178	}
179	) # pragma: no cover
180	return None	×
181
182
183	def administration_csv_delete(id: int):	1✔
184	test = settings.TEST_ENV	1✔
185	filename = "{0}-administration.csv".format(	1✔
186	"test" if test else COUNTRY_NAME
187	)
188	filepath = f"{STORAGE_PATH}/master_data/{filename}"	1✔
189	if os.path.exists(filepath):	1✔
190	df = pd.read_csv(filepath)	1✔
191	ix = find_index_by_id(df=df, id=id)	1✔
192	if ix is not None:	1✔
193	df.drop(index=ix, inplace=True)	1✔
194	df.to_csv(filepath, index=False)	1✔
195	return filepath	1✔
196	else:
197	logger.error(
198	{
199	"context": "delete_administration_row_csv",
200	"message": f"{filename} doesn't exist",
201	}
202	) # pragma: no cover
203	return None	×

akvo / akvo-mis / #637

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous