test: for db rename, change_column_type, describe

frappe/tests/test_db.py

@@ -12,6 +12,8 @@ from frappe.custom.doctype.custom_field.custom_field import create_custom_field
 from frappe.utils import random_string
 from frappe.utils.testutils import clear_custom_fields
 
+from .test_query_builder import CheckDB, db_type
+
 
 class TestDB(unittest.TestCase):
 	def test_get_value(self):
@@ -146,7 +148,7 @@ class TestDB(unittest.TestCase):
 
 		# Create documents under that doctype and query them via ORM
 		for _ in range(10):
-			docfields = { key.lower(): random_string(10) for key in fields }
+			docfields = {key.lower(): random_string(10) for key in fields}
 			doc = frappe.get_doc({"doctype": test_doctype, "description": random_string(20), **docfields})
 			doc.insert()
 			created_docs.append(doc.name)
@@ -189,3 +191,98 @@ class TestDB(unittest.TestCase):
 		for doc in created_docs:
 			frappe.delete_doc(test_doctype, doc)
 		clear_custom_fields(test_doctype)
+
+@CheckDB(db_type.MARIADB)
+class TestDDLCommandsMaria(unittest.TestCase):
+	test_table_name = "TestNotes"
+
+	def setUp(self) -> None:
+		frappe.db.commit()
+		frappe.db.sql(
+			f"""
+			CREATE TABLE `tab{self.test_table_name}` (`id` INT NULL,PRIMARY KEY (`id`));
+			"""
+		)
+
+	def tearDown(self) -> None:
+		frappe.db.sql(f"DROP TABLE tab{self.test_table_name};")
+		self.test_table_name = "TestNotes"
+
+	def test_rename(self) -> None:
+		new_table_name = f"{self.test_table_name}_new"
+		frappe.db.rename_table(self.test_table_name, new_table_name)
+		check_exists = frappe.db.sql(
+			f"""
+			SELECT * FROM INFORMATION_SCHEMA.TABLES
+			WHERE TABLE_NAME = N'tab{new_table_name}';
+			"""
+		)
+		self.assertGreater(len(check_exists), 0)
+		self.assertIn(f"tab{new_table_name}", check_exists[0])
+
+		# * so this table is deleted after the rename
+		self.test_table_name = new_table_name
+
+	def test_describe(self) -> None:
+		self.assertEqual(
+			(("id", "int(11)", "NO", "PRI", None, ""),),
+			frappe.db.DESC(self.test_table_name),
+		)
+
+	def test_change_type(self) -> None:
+		frappe.db.change_column_type("TestNotes", "id", "varchar(255)")
+		test_table_description = frappe.db.sql(f"DESC tab{self.test_table_name};")
+		self.assertGreater(len(test_table_description), 0)
+		self.assertIn("varchar(255)", test_table_description[0])
+
+
+@CheckDB(dbtype=db_type.POSTGRES)
+class TestDDLCommandsPost(unittest.TestCase):
+	test_table_name = "TestNotes"
+
+	def setUp(self) -> None:
+		frappe.db.sql(
+			f"""
+			CREATE TABLE "tab{self.test_table_name}" ("id" INT NULL,PRIMARY KEY ("id"))
+			"""
+		)
+
+	def tearDown(self) -> None:
+		frappe.db.sql(f'DROP TABLE "tab{self.test_table_name}"')
+		self.test_table_name = "TestNotes"
+
+	def test_rename(self) -> None:
+		new_table_name = f"{self.test_table_name}_new"
+		frappe.db.rename_table(self.test_table_name, new_table_name)
+		check_exists = frappe.db.sql(
+			f"""
+			SELECT EXISTS (
+			SELECT FROM information_schema.tables
+			WHERE  table_name = 'tab{new_table_name}'
+			);
+			"""
+		)
+		self.assertTrue(check_exists[0][0])
+
+		# * so this table is deleted after the rename
+		self.test_table_name = new_table_name
+
+	def test_describe(self) -> None:
+		self.assertEqual([("id",)], frappe.db.DESC(self.test_table_name))
+
+	def test_change_type(self) -> None:
+		frappe.db.change_column_type(self.test_table_name, "id", "varchar(255)")
+		check_change = frappe.db.sql(
+			f"""
+			SELECT
+				table_name,
+				column_name,
+				data_type
+			FROM
+				information_schema.columns
+			WHERE
+				table_name = 'tab{self.test_table_name}'
+			"""
+		)
+		self.assertGreater(len(check_change), 0)
+		self.assertIn("character varying", check_change[0])