vassili/seitime-frappe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 4

Merge pull request #34560 from AarDG10/setup-postgres-ci

Browse source 
ci: Setup Postgres test environment
This commit is contained in:
Ankush Menat 2025-11-14 11:30:07 +05:30 • committed by GitHub
commit 02a05c86bc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 178 additions and 107 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
.github/actions/setup/action.yml vendored
View file

@ -101,9 +101,15 @@ runs:
# Install System Dependencies
start_time=$(date +%s)
curl -LsS https://r.mariadb.com/downloads/mariadb_repo_setup | sudo bash
sudo sh -c 'echo "deb http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list'
wget --quiet -O - https://www.postgresql.org/media/keys/ACCC4CF8.asc | sudo apt-key add -
sudo apt -qq update
sudo apt -qq remove mysql-server mysql-client
sudo apt -qq install libcups2-dev redis-server mariadb-client libmariadb-dev
sudo apt -qq remove -y postgresql-client postgresql-client-16 postgresql-client-common
sudo apt -qq install libcups2-dev redis-server mariadb-client libmariadb-dev postgresql-client-18 libpq-dev
echo "/usr/lib/postgresql/18/bin" >> $GITHUB_PATH
wget -q -O /tmp/wkhtmltox.deb https://github.com/wkhtmltopdf/packaging/releases/download/0.12.6.1-2/wkhtmltox_0.12.6.1-2.jammy_amd64.deb
sudo apt install /tmp/wkhtmltox.deb
@ -169,6 +175,14 @@ runs:
mariadb --host 127.0.0.1 --port 3306 -u root -p${{ inputs.db-root-password }} -e "FLUSH PRIVILEGES";
fi
if [ "$DB" == "postgres" ]; then
export PGPASSWORD='travis'
psql -h 127.0.0.1 -p 5432 -c "CREATE DATABASE test_frappe" -U postgres
psql -h 127.0.0.1 -p 5432 -c "CREATE USER test_frappe WITH PASSWORD 'test_frappe'" -U postgres
psql -h 127.0.0.1 -p 5432 -U postgres -c "GRANT ALL PRIVILEGES ON DATABASE test_frappe TO test_frappe;"
unset PGPASSWORD
fi
- shell: bash -e {0}
run: |
# Install App(s)

18
.github/helper/db/postgres.json vendored Normal file
View file

@ -0,0 +1,18 @@
{
"db_host": "127.0.0.1",
"db_port": 5432,
"db_name": "test_frappe",
"db_password": "test_frappe",
"db_type": "postgres",
"allow_tests": true,
"auto_email_id": "test@example.com",
"mail_server": "localhost",
"mail_port": 2525,
"mail_login": "test@example.com",
"mail_password": "test",
"admin_password": "admin",
"root_login": "postgres",
"root_password": "travis",
"host_name": "http://test_site:8000",
"server_script_enabled": true
}

17
.github/workflows/_base-server-tests.yml vendored
View file

@ -22,6 +22,10 @@ on:
required: false
type: boolean
default: false
enable-postgres:
required: false
type: boolean
default: true
enable-coverage:
required: false
type: boolean
@ -62,9 +66,20 @@ jobs:
strategy:
fail-fast: false
matrix:
db: ${{ fromJson(inputs.enable-sqlite && '["mariadb", "sqlite"]' || '["mariadb"]') }}
db: ${{ fromJson(inputs.enable-sqlite && (inputs.enable-postgres && '["mariadb", "postgres", "sqlite"]' || '["mariadb", "sqlite"]') || (inputs.enable-postgres && '["mariadb", "postgres"]' || '["mariadb"]')) }}
index: ${{ fromJson(needs.gen-idx-integration.outputs.indices) }}
services:
postgres:
image: postgres:18.0
ports:
- 5432:5432
env:
POSTGRES_PASSWORD: travis
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 3
mariadb:
image: mariadb:11.8
ports:

1
.github/workflows/server-tests.yml vendored
View file

@ -44,6 +44,7 @@ jobs:
name: Tests
uses: ./.github/workflows/_base-server-tests.yml
with:
enable-postgres: ${{ contains(github.event.pull_request.labels.*.name, 'postgres') }} # This enables PostgreSQL to run tests
enable-sqlite: false # This will test against both MariaDB and SQLite if enabled
parallel-runs: 2
enable-coverage: ${{ github.event_name != 'pull_request' }}

13
frappe/commands/test_commands.py
View file

@ -469,11 +469,10 @@ class TestCommands(BaseTestCommands):
self.assertEqual(check_password("Administrator", original_password), "Administrator")
@skipIf(
not (frappe.conf.root_password and frappe.conf.admin_password and frappe.conf.db_type == "mariadb"),
not (frappe.conf.root_password and frappe.conf.admin_password and frappe.conf.db_type != "sqlite"),
"DB Root password and Admin password not set in config",
)
def test_bench_drop_site_should_archive_site(self):
# TODO: Make this test postgres compatible
site = TEST_SITE
self.execute(
@ -499,7 +498,7 @@ class TestCommands(BaseTestCommands):
self.assertTrue(os.path.exists(archive_directory))
@skipIf(
not (frappe.conf.root_password and frappe.conf.admin_password and frappe.conf.db_type == "mariadb"),
not (frappe.conf.root_password and frappe.conf.admin_password and frappe.conf.db_type != "sqlite"),
"DB Root password and Admin password not set in config",
)
def test_force_install_app(self):
@ -656,10 +655,10 @@ class TestBackups(BaseTestCommands):
except OSError:
pass
@run_only_if(db_type_is.MARIADB)
def test_backup_no_options(self):
"""Take a backup without any options"""
before_backup = fetch_latest_backups(partial=True)
time.sleep(1)
self.execute("bench --site {site} backup")
after_backup = fetch_latest_backups(partial=True)
@ -1003,9 +1002,11 @@ class TestDBCli(BaseTestCommands):
self.execute("bench --site {site} db-console", kwargs={"cmd_input": cmd_input})
self.assertEqual(self.returncode, 0)
@run_only_if(db_type_is.MARIADB)
def test_db_cli_with_sql(self):
self.execute("bench --site {site} db-console -e 'select 1'")
if frappe.db.db_type == "postgres":
self.execute("bench --site {site} db-console -c 'select 1'")
elif frappe.db.db_type == "mariadb":
self.execute("bench --site {site} db-console -e 'select 1'")
self.assertEqual(self.returncode, 0)
self.assertIn("1", self.stdout)

15
frappe/core/doctype/doctype/test_doctype.py
View file

@ -840,7 +840,20 @@ class TestDocType(IntegrationTestCase):
],
).insert(ignore_if_duplicate=True)
decimal_field_type = frappe.db.get_column_type(doctype.name, "decimal_field")
self.assertIn("(30,3)", decimal_field_type.lower())
if frappe.db.db_type == "postgres":
result = frappe.db.sql(
"""
SELECT numeric_precision, numeric_scale
FROM information_schema.columns
WHERE lower(table_name) = lower(%s)
AND column_name = %s
""",
(f"tab{doctype.name}", "decimal_field"),
)
length, precision = result[0]
self.assertEqual((length, precision), (30, 3))
elif frappe.db.db_type == "mariadb":
self.assertIn("(30,3)", decimal_field_type.lower())
def test_decimal_field_precision_exceeds_length(self):
doctype = new_doctype(

9
frappe/core/doctype/user_invitation/test_user_invitation.py
View file

@ -55,15 +55,18 @@ class IntegrationTestUserInvitation(IntegrationTestCase):
@classmethod
def delete_all_user_roles(cls):
frappe.db.sql("DELETE FROM `tabUser Role`")
query = "DELETE FROM `tabUser Role`"
frappe.db.sql(cls.normalize_sql(query))
@classmethod
def delete_all_invitations(cls):
frappe.db.sql("DELETE FROM `tabUser Invitation`")
query = "DELETE FROM `tabUser Invitation`"
frappe.db.sql(cls.normalize_sql(query))
@classmethod
def delete_invitation(cls, name: str):
frappe.db.sql(f'DELETE FROM `tabUser Invitation` WHERE name = "{name}"')
query = "DELETE FROM `tabUser Invitation` WHERE name = %s"
frappe.db.sql(cls.normalize_sql(query), name)
def setUp(self):
super().setUp()

1
frappe/desk/reportview.py
View file

@ -63,6 +63,7 @@ def get_count() -> int | None:
distinct = "distinct " if args.distinct else ""
args.limit = cint(args.limit)
fieldname = f"{distinct}`tab{args.doctype}`.name"
args.pop("distinct") # to avoid a double DISTINCT concat in db_query
args.order_by = None
# args.limit is specified to avoid getting accurate count.

7
frappe/email/test_smtp.py
View file

@ -26,7 +26,8 @@ class TestSMTP(IntegrationTestCase):
# remove mail_server config so that test@example.com is not created
mail_server = frappe.conf.get("mail_server")
del frappe.conf["mail_server"]
if "mail_server" in frappe.conf:
del frappe.conf["mail_server"]
frappe.local.outgoing_email_account = {}
@ -47,9 +48,9 @@ class TestSMTP(IntegrationTestCase):
password="password",
enable_outgoing=1,
default_outgoing=1,
append_to="Todo",
append_to="ToDo",
)
self.assertEqual(EmailAccount.find_outgoing(match_by_doctype="Todo").email_id, "append_to@gmail.com")
self.assertEqual(EmailAccount.find_outgoing(match_by_doctype="ToDo").email_id, "append_to@gmail.com")
# add back the mail_server
frappe.conf["mail_server"] = mail_server

2
frappe/tests/classes/integration_test_case.py
View file

@ -118,7 +118,7 @@ class IntegrationTestCase(UnitTestCase):
def _sql_with_count(*args, **kwargs):
ret = orig_sql(*args, **kwargs)
queries.append(args[0].last_query)
queries.append(str(args[0].last_query))
return ret
try:

2
frappe/tests/classes/unit_test_case.py
View file

@ -105,6 +105,8 @@ class UnitTestCase(unittest.TestCase, BaseTestCase):
"""Formats SQL consistently so simple string comparisons can work on them."""
import sqlparse
if frappe.db.db_type == "postgres":
query = query.replace("`", '"')
return sqlparse.format(query.strip(), keyword_case="upper", reindent=True, strip_comments=True)

15
frappe/tests/test_db.py
View file

@ -1050,14 +1050,13 @@ class TestDDLCommandsPost(IntegrationTestCase):
def test_is(self):
user = frappe.qb.DocType("User")
self.assertIn(
'coalesce("name",',
frappe.db.get_values(user, filters={user.name: ("is", "set")}, run=False).lower(),
)
self.assertIn(
'coalesce("name",',
frappe.db.get_values(user, filters={user.name: ("is", "not set")}, run=False).lower(),
)
query_is_set = frappe.db.get_values(user, filters={user.name: ("is", "set")}, run=False).lower()
query_is_not_set = frappe.db.get_values(
user, filters={user.name: ("is", "not set")}, run=False
).lower()
self.assertIn('"name"<>%(param1)s', query_is_set)
self.assertIn('"name" is null or "name"=%(param1)s', query_is_not_set)
@run_only_if(db_type_is.POSTGRES)

7
frappe/tests/test_db_query.py
View file

@ -221,8 +221,8 @@ class TestDBQuery(IntegrationTestCase):
# get as conditions
if frappe.db.db_type == "mariadb":
assertion_string = """(((ifnull(`tabTest Blog Post`.`name`, '')='' or `tabTest Blog Post`.`name` in ('_Test Blog Post 1', '_Test Blog Post'))))"""
else:
assertion_string = """(((ifnull(cast(`tabBlog Post`.`name` as varchar), '')='' or cast(`tabBlog Post`.`name` as varchar) in ('_Test Blog Post 1', '_Test Blog Post'))))"""
elif frappe.db.db_type == "postgres":
assertion_string = """(((ifnull(cast(`tabTest Blog Post`.`name` as varchar), '')='' or cast(`tabTest Blog Post`.`name` as varchar) in ('_Test Blog Post 1', '_Test Blog Post'))))"""
self.assertEqual(build_match_conditions(as_condition=True), assertion_string)
@ -1244,7 +1244,6 @@ class TestDBQuery(IntegrationTestCase):
class TestReportView(IntegrationTestCase):
@run_only_if(db_type_is.MARIADB) # TODO: postgres name casting is messed up
def test_get_count(self):
frappe.local.request = frappe._dict()
frappe.local.request.method = "GET"
@ -1282,7 +1281,7 @@ class TestReportView(IntegrationTestCase):
child_filter_response = execute_cmd("frappe.desk.reportview.get_count")
current_value = frappe.db.sql(
# the below query is equivalent to the one in reportview.get_count
"select distinct count(distinct `tabDocType`.name) as total_count"
"select count(distinct `tabDocType`.name) as total_count"
" from `tabDocType` left join `tabDocField`"
" on (`tabDocField`.parenttype = 'DocType' and `tabDocField`.parent = `tabDocType`.name)"
" where `tabDocField`.`fieldtype` = 'Data'"

70
frappe/tests/test_db_update.py
View file

@ -93,10 +93,29 @@ class TestDBUpdate(IntegrationTestCase):
self.assertEqual(email_sig_column.index, 1)
def check_unique_indexes(self, doctype: str, field: str):
indexes = frappe.db.sql(
f"""show index from `tab{doctype}` where column_name = '{field}' and Non_unique = 0""",
as_dict=1,
)
if frappe.db.db_type == "postgres":
"""Check if the column has a unique index (PostgreSQL equivalent of "SHOW INDEX ... WHERE Non_unique = 0")"""
indexes = frappe.db.sql(
"""
SELECT i.relname AS index_name, a.attname AS column_name
FROM
pg_class t
JOIN pg_index ix ON t.oid = ix.indrelid
JOIN pg_class i ON i.oid = ix.indexrelid
JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = ANY(ix.indkey)
WHERE
t.relname = %s
AND a.attname = %s
AND ix.indisunique = true
""",
(f"tab{doctype}", field),
as_dict=1,
)
elif frappe.db.db_type == "mariadb":
indexes = frappe.db.sql(
f"""show index from `tab{doctype}` where column_name = '{field}' and Non_unique = 0""",
as_dict=1,
)
self.assertEqual(
len(indexes), 1, msg=f"There should be 1 index on {doctype}.{field}, found {indexes}"
)
@ -111,7 +130,6 @@ class TestDBUpdate(IntegrationTestCase):
doctype.save()
frappe.get_doc(doctype=doctype.name, int_field=2**62 - 1).insert()
@run_only_if(db_type_is.MARIADB)
def test_unique_index_on_install(self):
"""Only one unique index should be added"""
for dt in frappe.get_all("DocType", {"is_virtual": 0, "issingle": 0}, pluck="name"):
@ -126,30 +144,31 @@ class TestDBUpdate(IntegrationTestCase):
"""Only one unique index should be added"""
doctype = new_doctype(unique=1).insert()
field = "some_fieldname"
try:
field = "some_fieldname"
self.check_unique_indexes(doctype.name, field)
doctype.fields[0].length = 142
doctype.save()
self.check_unique_indexes(doctype.name, field)
self.check_unique_indexes(doctype.name, field)
doctype.fields[0].length = 142
doctype.save()
self.check_unique_indexes(doctype.name, field)
doctype.fields[0].unique = 0
doctype.save()
doctype.fields[0].unique = 0
doctype.save()
doctype.fields[0].unique = 1
doctype.save()
self.check_unique_indexes(doctype.name, field)
doctype.fields[0].unique = 1
doctype.save()
self.check_unique_indexes(doctype.name, field)
# New column with a unique index
# This works because index name is same as fieldname.
new_field = frappe.copy_doc(doctype.fields[0])
new_field.fieldname = "duplicate_field"
doctype.append("fields", new_field)
doctype.save()
self.check_unique_indexes(doctype.name, new_field.fieldname)
doctype.delete()
frappe.db.commit()
# New column with a unique index
# This works because index name is same as fieldname.
new_field = frappe.copy_doc(doctype.fields[0])
new_field.fieldname = "duplicate_field"
doctype.append("fields", new_field)
doctype.save()
self.check_unique_indexes(doctype.name, new_field.fieldname)
finally:
doctype.delete()
frappe.db.commit()
def test_uuid_varchar_migration(self):
doctype = new_doctype().insert()
@ -176,7 +195,6 @@ class TestDBUpdate(IntegrationTestCase):
self.assertEqual(frappe.db.get_column_type(referring_doctype.name, link), "uuid")
@run_only_if(db_type_is.MARIADB)
def test_varchar_length(self):
from frappe.database.schema import add_column

1
frappe/tests/test_email.py
View file

@ -371,7 +371,6 @@ class TestEmailIntegrationTest(IntegrationTestCase):
self.assertEqual(sent_mail["subject"], subject)
self.assertSetEqual(set(recipients.split(",")), {m["to"][0] for m in sent_mails})
@run_only_if(db_type_is.MARIADB)
@IntegrationTestCase.change_settings("System Settings", store_attached_pdf_document=1)
def test_store_attachments(self):
""" "attach print" feature just tells email queue which document to attach, this is not

91
frappe/tests/test_query.py
View file

@ -67,9 +67,8 @@ class TestQuery(IntegrationTestCase):
def setUp(self):
setup_for_tests()
@run_only_if(db_type_is.MARIADB)
def test_multiple_tables_in_filters(self):
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
["*"],
@ -81,7 +80,6 @@ class TestQuery(IntegrationTestCase):
"SELECT `tabDocType`.* FROM `tabDocType` LEFT JOIN `tabDocField` ON `tabDocField`.`parent`=`tabDocType`.`name` AND `tabDocField`.`parenttype`='DocType' AND `tabDocField`.`parentfield`='fields' WHERE `tabDocField`.`name` LIKE 'f%' AND `tabDocType`.`parent`='something'",
)
@run_only_if(db_type_is.MARIADB)
def test_string_fields(self):
self.assertEqual(
frappe.qb.get_query("User", fields="name, email", filters={"name": "Administrator"}).get_sql(),
@ -352,88 +350,73 @@ class TestQuery(IntegrationTestCase):
.get_sql(),
)
@run_only_if(db_type_is.MARIADB)
def test_filters(self):
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"module.app_name": "frappe"},
).get_sql(),
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module` WHERE `tabModule Def`.`app_name`='frappe'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module` WHERE `tabModule Def`.`app_name`='frappe'",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"module.app_name": ("like", "frap%")},
).get_sql(),
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module` WHERE `tabModule Def`.`app_name` LIKE 'frap%'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module` WHERE `tabModule Def`.`app_name` LIKE 'frap%'",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"permissions.role": "System Manager"},
).get_sql(),
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabDocPerm` ON `tabDocPerm`.`parent`=`tabDocType`.`name` AND `tabDocPerm`.`parenttype`='DocType' AND `tabDocPerm`.`parentfield`='permissions' WHERE `tabDocPerm`.`role`='System Manager'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabDocPerm` ON `tabDocPerm`.`parent`=`tabDocType`.`name` AND `tabDocPerm`.`parenttype`='DocType' AND `tabDocPerm`.`parentfield`='permissions' WHERE `tabDocPerm`.`role`='System Manager'",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["module"],
filters="",
).get_sql(),
"SELECT `module` FROM `tabDocType` WHERE `name`=''".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `module` FROM `tabDocType` WHERE `name`=''",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
filters=["ToDo", "Note"],
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name` IN ('ToDo','Note')".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name` IN ('ToDo','Note')",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
filters={"name": ("in", [])},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name` IN ('')".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name` IN ('')",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
filters=[1, 2, 3],
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name` IN (1,2,3)".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name` IN (1,2,3)",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
filters=[],
).get_sql(),
"SELECT `name` FROM `tabDocType`".replace("`", '"' if frappe.db.db_type == "postgres" else "`"),
"SELECT `name` FROM `tabDocType`",
)
def test_nested_filters(self):
@ -1191,7 +1174,7 @@ class TestQuery(IntegrationTestCase):
query = frappe.qb.get_query(
"DocType",
fields=["module.app_name", "name"],
group_by="module.app_name",
group_by="module.app_name, name",
)
result = query.run(as_dict=True)
self.assertTrue(len(result) > 0)
@ -1210,7 +1193,7 @@ class TestQuery(IntegrationTestCase):
"Note",
fields=["seen_by.user", "name"],
filters={"name": note.name},
group_by="seen_by.user",
group_by="seen_by.user, name",
)
result = query.run(as_dict=True)
self.assertTrue(len(result) >= 1)
@ -1268,7 +1251,7 @@ class TestQuery(IntegrationTestCase):
query = frappe.qb.get_query(
"DocType",
fields=["module", "module.app_name", "name"],
group_by="module, module.app_name",
group_by="module, module.app_name, name",
order_by="module.app_name",
)
result = query.run(as_dict=True)
@ -1513,7 +1496,7 @@ class TestQuery(IntegrationTestCase):
# Test simple function without alias
query = frappe.qb.get_query("User", fields=["user_type", {"COUNT": "name"}], group_by="user_type")
sql = query.get_sql()
self.assertIn("COUNT(`name`)", sql)
self.assertIn(self.normalize_sql("COUNT(`name`)"), sql)
self.assertIn("GROUP BY", sql)
# Test function with alias
@ -1521,52 +1504,54 @@ class TestQuery(IntegrationTestCase):
"User", fields=[{"COUNT": "name", "as": "total_users"}], group_by="user_type"
)
sql = query.get_sql()
self.assertIn("COUNT(`name`) `total_users`", sql)
self.assertIn(self.normalize_sql("COUNT(`name`) `total_users`"), sql)
# Test SUM function with alias
query = frappe.qb.get_query(
"User", fields=[{"SUM": "enabled", "as": "total_enabled"}], group_by="user_type"
)
sql = query.get_sql()
self.assertIn("SUM(`enabled`) `total_enabled`", sql)
self.assertIn(self.normalize_sql("SUM(`enabled`) `total_enabled`"), sql)
# Test MAX function
query = frappe.qb.get_query(
"User", fields=[{"MAX": "creation", "as": "latest_user"}], group_by="user_type"
)
sql = query.get_sql()
self.assertIn("MAX(`creation`) `latest_user`", sql)
self.assertIn(self.normalize_sql("MAX(`creation`) `latest_user`"), sql)
# Test MIN function
query = frappe.qb.get_query(
"User", fields=[{"MIN": "creation", "as": "earliest_user"}], group_by="user_type"
)
sql = query.get_sql()
self.assertIn("MIN(`creation`) `earliest_user`", sql)
self.assertIn(self.normalize_sql("MIN(`creation`) `earliest_user`"), sql)
# Test AVG function
query = frappe.qb.get_query(
"User", fields=[{"AVG": "enabled", "as": "avg_enabled"}], group_by="user_type"
)
sql = query.get_sql()
self.assertIn("AVG(`enabled`) `avg_enabled`", sql)
self.assertIn(self.normalize_sql("AVG(`enabled`) `avg_enabled`"), sql)
# Test ABS function
query = frappe.qb.get_query("User", fields=[{"ABS": "enabled", "as": "abs_enabled"}])
sql = query.get_sql()
self.assertIn("ABS(`enabled`) `abs_enabled`", sql)
self.assertIn(self.normalize_sql("ABS(`enabled`) `abs_enabled`"), sql)
# Test IFNULL function with two parameters
query = frappe.qb.get_query(
"User", fields=[{"IFNULL": ["first_name", "'Unknown'"], "as": "safe_name"}]
)
sql = query.get_sql()
self.assertIn("IFNULL(`first_name`,'Unknown') `safe_name`", sql)
self.assertIn(
self.normalize_sql("IFNULL(`first_name`,'Unknown') `safe_name`"), self.normalize_sql(sql)
)
# Test TIMESTAMP function
query = frappe.qb.get_query("User", fields=[{"TIMESTAMP": "creation", "as": "ts"}])
sql = query.get_sql()
self.assertIn("TIMESTAMP(`creation`) `ts`", sql)
self.assertIn(self.normalize_sql("TIMESTAMP(`creation`) `ts`"), self.normalize_sql(sql))
# Test mixed regular fields and function fields
query = frappe.qb.get_query(
@ -1579,21 +1564,23 @@ class TestQuery(IntegrationTestCase):
group_by="user_type",
)
sql = query.get_sql()
self.assertIn("`user_type`", sql)
self.assertIn("COUNT(`name`) `total_users`", sql)
self.assertIn("MAX(`creation`) `latest_creation`", sql)
self.assertIn(self.normalize_sql("`user_type`"), sql)
self.assertIn(self.normalize_sql("COUNT(`name`) `total_users`"), sql)
self.assertIn(self.normalize_sql("MAX(`creation`) `latest_creation`"), sql)
# Test NOW function with no arguments
query = frappe.qb.get_query("User", fields=[{"NOW": None, "as": "current_time"}])
sql = query.get_sql()
self.assertIn("NOW() `current_time`", sql)
self.assertIn(self.normalize_sql("NOW() `current_time`"), sql)
# Test CONCAT function (which is supported)
query = frappe.qb.get_query(
"User", fields=[{"CONCAT": ["first_name", "last_name"], "as": "full_name"}]
)
sql = query.get_sql()
self.assertIn("CONCAT(`first_name`,`last_name`) `full_name`", sql)
self.assertIn(
self.normalize_sql("CONCAT(`first_name`,`last_name`) `full_name`"), self.normalize_sql(sql)
)
# Test unsupported function validation
with self.assertRaises(frappe.ValidationError) as cm:
@ -1611,7 +1598,7 @@ class TestQuery(IntegrationTestCase):
self.assertIn("Unsupported function or invalid field name: DROP", str(cm.exception))
def test_not_equal_condition_on_none(self):
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
["*"],