diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-18 05:31:44 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-18 05:31:44 +0000 |
commit | f77392695c09f9fef9386c112aef0e2b2f6fcd1a (patch) | |
tree | 04c428a7cfc9b7c6dbea73b5697f8c201d9106ff /tests/test_specials.py | |
parent | Initial commit. (diff) | |
download | python-pgspecial-f77392695c09f9fef9386c112aef0e2b2f6fcd1a.tar.xz python-pgspecial-f77392695c09f9fef9386c112aef0e2b2f6fcd1a.zip |
Adding upstream version 2.1.2.upstream/2.1.2upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rwxr-xr-x | tests/test_specials.py | 1143 |
1 files changed, 1143 insertions, 0 deletions
diff --git a/tests/test_specials.py b/tests/test_specials.py new file mode 100755 index 0000000..08319f4 --- /dev/null +++ b/tests/test_specials.py @@ -0,0 +1,1143 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +import pytest +from dbutils import dbtest, POSTGRES_USER, SERVER_VERSION, foreign_db_environ, fdw_test +import itertools +import locale + +objects_listing_headers = ["Schema", "Name", "Type", "Owner", "Size", "Description"] + +# note: technically, this is the database encoding, not the client +# locale but for the purpose of testing we can assume the server +# and the client are using the same locale +# note 2: locale.getlocale() does not return the raw values, +# in particular it transforms C into en_US, so we use .setlocale() +# instead as that matches the C library function +LC_COLLATE = locale.setlocale(locale.LC_COLLATE, None) +LC_CTYPE = locale.setlocale(locale.LC_CTYPE, None) + + +@dbtest +def test_slash_l(executor): + results = executor(r"\l") + row = ("_test_db", "postgres", "UTF8", LC_COLLATE, LC_CTYPE, None) + headers = ["Name", "Owner", "Encoding", "Collate", "Ctype", "Access privileges"] + assert row in results[1] + assert headers == results[2] + + +@dbtest +def test_slash_l_pattern(executor): + results = executor(r"\l _test*") + row = [("_test_db", "postgres", "UTF8", LC_COLLATE, LC_CTYPE, None)] + headers = ["Name", "Owner", "Encoding", "Collate", "Ctype", "Access privileges"] + assert row == results[1] + assert headers == results[2] + + +@dbtest +def test_slash_l_verbose(executor): + results = executor(r"\l+") + headers = [ + "Name", + "Owner", + "Encoding", + "Collate", + "Ctype", + "Access privileges", + "Size", + "Tablespace", + "Description", + ] + assert headers == results[2] + + +@dbtest +def test_slash_du(executor): + results = executor(r"\du") + row = ("postgres", True, True, True, True, True, -1, None, [], True) + headers = [ + "rolname", + "rolsuper", + "rolinherit", + "rolcreaterole", + "rolcreatedb", + "rolcanlogin", + "rolconnlimit", + "rolvaliduntil", + "memberof", + "rolreplication", + ] + assert headers == results[2] + assert row in results[1] + + +@dbtest +def test_slash_du_pattern(executor): + results = executor(r"\du post*") + row = [("postgres", True, True, True, True, True, -1, None, [], True)] + headers = [ + "rolname", + "rolsuper", + "rolinherit", + "rolcreaterole", + "rolcreatedb", + "rolcanlogin", + "rolconnlimit", + "rolvaliduntil", + "memberof", + "rolreplication", + ] + assert headers == results[2] + assert row == results[1] + + +@dbtest +def test_slash_du_verbose(executor): + results = executor(r"\du+") + row = ("postgres", True, True, True, True, True, -1, None, [], None, True) + headers = [ + "rolname", + "rolsuper", + "rolinherit", + "rolcreaterole", + "rolcreatedb", + "rolcanlogin", + "rolconnlimit", + "rolvaliduntil", + "memberof", + "description", + "rolreplication", + ] + assert headers == results[2] + assert row in results[1] + + +@dbtest +def test_slash_d(executor): + results = executor(r"\d") + title = None + rows = [ + ("public", "Inh1", "table", POSTGRES_USER), + ("public", "inh2", "table", POSTGRES_USER), + ("public", "mvw1", "materialized view", POSTGRES_USER), + ("public", "tbl1", "table", POSTGRES_USER), + ("public", "tbl2", "table", POSTGRES_USER), + ("public", "tbl2_id2_seq", "sequence", POSTGRES_USER), + ("public", "tbl3", "table", POSTGRES_USER), + ("public", "vw1", "view", POSTGRES_USER), + ] + headers = objects_listing_headers[:-2] + status = "SELECT 8" + expected = [title, rows, headers, status] + + assert results == expected + + +@dbtest +def test_slash_d_verbose(executor): + results = executor(r"\d+") + title = None + rows = [ + ("public", "Inh1", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "inh2", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "mvw1", "materialized view", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl1", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl2", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl2_id2_seq", "sequence", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl3", "table", POSTGRES_USER, "0 bytes", None), + ("public", "vw1", "view", POSTGRES_USER, "0 bytes", None), + ] + headers = objects_listing_headers + status = "SELECT 8" + expected = [title, rows, headers, status] + + assert results == expected + + +@dbtest +def test_slash_d_table_1(executor): + results = executor(r"\d tbl1") + title = None + rows = [ + ["id1", "integer", " not null"], + ["txt1", "text", " not null"], + ] + headers = ["Column", "Type", "Modifiers"] + status = ( + 'Indexes:\n "id_text" PRIMARY KEY, btree (id1, txt1)\n' + "Number of child tables: 2 (Use \\d+ to list them.)\n" + ) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_d_table_2(executor): + results = executor(r"\d tbl2") + title = None + rows = [ + ["id2", "integer", " not null default nextval('tbl2_id2_seq'::regclass)"], + ["txt2", "text", ""], + ] + headers = ["Column", "Type", "Modifiers"] + status = "Number of child tables: 1 (Use \\d+ to list them.)\n" + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_d_test_generated_default(executor): + results = executor(r"\d schema3.test_generated_default") + headers = ["Column", "Type", "Modifiers"] + status = 'Indexes:\n "test_generated_default_pkey" PRIMARY KEY, btree (id)\n' + rows = [ + ["id", "integer", " not null generated by default as identity"], + ["some_stuff", "text", ""], + ] + assert rows == results[1] + assert headers == results[2] + assert status == results[3] + + +@dbtest +def test_slash_d_table_verbose_1(executor): + title = None + headers = ["Column", "Type", "Modifiers", "Storage", "Stats target", "Description"] + + results = executor(r"\d+ tbl1") + rows = [ + ["id1", "integer", " not null", "plain", None, None], + ["txt1", "text", " not null", "extended", None, None], + ] + status = ( + 'Indexes:\n "id_text" PRIMARY KEY, btree (id1, txt1)\n' + 'Child tables: "Inh1",\n' + " inh2\n" + "Has OIDs: no\n" + ) + expected = [title, rows, headers, status] + assert results == expected + + results = executor(r'\d+ "Inh1"') + rows = [ + ["id1", "integer", " not null", "plain", None, None], + ["txt1", "text", " not null", "extended", None, None], + ["value1", "integer", "", "plain", None, None], + ] + status = "Inherits: tbl1\n" "Has OIDs: no\n" + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_d_table_verbose_2(executor): + title = None + headers = ["Column", "Type", "Modifiers", "Storage", "Stats target", "Description"] + + results = executor(r"\d+ tbl2") + rows = [ + [ + "id2", + "integer", + " not null default nextval('tbl2_id2_seq'::regclass)", + "plain", + None, + None, + ], + ["txt2", "text", "", "extended", None, None], + ] + status = "Child tables: inh2\n" "Has OIDs: no\n" + expected = [title, rows, headers, status] + assert results == expected + + results = executor(r"\d+ inh2") + rows = [ + ["id1", "integer", " not null", "plain", None, None], + ["txt1", "text", " not null", "extended", None, None], + [ + "id2", + "integer", + " not null default nextval('tbl2_id2_seq'::regclass)", + "plain", + None, + None, + ], + ["txt2", "text", "", "extended", None, None], + ["value2", "integer", "", "plain", None, None], + ] + status = "Inherits: tbl1,\n" " tbl2\n" "Has OIDs: no\n" + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_d_view_verbose(executor): + title = None + headers = ["Column", "Type", "Modifiers", "Storage", "Description"] + + results = executor(r"\d+ vw1") + rows = [ + ["id1", "integer", "", "plain", None], + ["txt1", "text", "", "extended", None], + ] + status = "View definition:\n SELECT tbl1.id1,\n tbl1.txt1\n FROM tbl1; \n" + + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_d_table_with_exclusion(executor): + results = executor(r"\d tbl3") + title = None + rows = [["c3", "circle", ""]] + headers = ["Column", "Type", "Modifiers"] + status = 'Indexes:\n "tbl3_c3_excl" EXCLUDE USING gist (c3 WITH &&)\n' + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_d_table_2_in_schema(executor): + results = executor(r"\d schema2.tbl2") + title = None + rows = [ + [ + "id2", + "integer", + " not null default nextval('schema2.tbl2_id2_seq'::regclass)", + ], + ["txt2", "text", ""], + ] + headers = ["Column", "Type", "Modifiers"] + status = "" + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dn(executor): + """List all schemas.""" + results = executor(r"\dn") + title = None + if SERVER_VERSION >= 150001: + rows = [ + ("public", "pg_database_owner"), + ("schema1", POSTGRES_USER), + ("schema2", POSTGRES_USER), + ("schema3", POSTGRES_USER), + ] + else: + rows = [ + ("public", POSTGRES_USER), + ("schema1", POSTGRES_USER), + ("schema2", POSTGRES_USER), + ("schema3", POSTGRES_USER), + ] + + headers = ["Name", "Owner"] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dp(executor): + """List all schemas.""" + results = executor(r"\dp") + title = None + rows = [ + ("public", "Inh1", "table", None, "", ""), + ("public", "inh2", "table", None, "", ""), + ("public", "mvw1", "materialized view", None, "", ""), + ("public", "tbl1", "table", None, "", ""), + ("public", "tbl2", "table", None, "", ""), + ("public", "tbl2_id2_seq", "sequence", None, "", ""), + ("public", "tbl3", "table", None, "", ""), + ("public", "vw1", "view", None, "", ""), + ] + + headers = [ + "Schema", + "Name", + "Type", + "Access privileges", + "Column privileges", + "Policies", + ] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dp_pattern_table(executor): + """List all schemas.""" + results = executor(r"\dp i*2") + title = None + rows = [("public", "inh2", "table", None, "", "")] + headers = [ + "Schema", + "Name", + "Type", + "Access privileges", + "Column privileges", + "Policies", + ] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dp_pattern_schema(executor): + """List all schemas.""" + results = executor(r"\dp schema2.*") + title = None + rows = [ + ("schema2", "tbl2", "table", None, "", ""), + ("schema2", "tbl2_id2_seq", "sequence", None, "", ""), + ] + headers = [ + "Schema", + "Name", + "Type", + "Access privileges", + "Column privileges", + "Policies", + ] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dp_pattern_alias(executor): + """List all schemas.""" + results = executor(r"\z i*2") + title = None + rows = [("public", "inh2", "table", None, "", "")] + headers = [ + "Schema", + "Name", + "Type", + "Access privileges", + "Column privileges", + "Policies", + ] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_ddp(executor): + """List all schemas.""" + results = executor(r"\ddp") + title = None + rows = [ + ("postgres", "schema1", "table", "test_role=r/postgres"), + ("postgres", "schema2", "table", "test_role=arwdDxt/postgres"), + ] + + headers = [ + "Owner", + "Schema", + "Type", + "Access privileges", + ] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_ddp_pattern(executor): + """List all schemas.""" + results = executor(r"\ddp schema2") + title = None + rows = [("postgres", "schema2", "table", "test_role=arwdDxt/postgres")] + headers = [ + "Owner", + "Schema", + "Type", + "Access privileges", + ] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dt(executor): + """List all tables in public schema.""" + results = executor(r"\dt") + title = None + rows = [ + ("public", "Inh1", "table", POSTGRES_USER), + ("public", "inh2", "table", POSTGRES_USER), + ("public", "tbl1", "table", POSTGRES_USER), + ("public", "tbl2", "table", POSTGRES_USER), + ("public", "tbl3", "table", POSTGRES_USER), + ] + headers = objects_listing_headers[:-2] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dt_verbose(executor): + """List all tables in public schema in verbose mode.""" + results = executor(r"\dt+") + title = None + rows = [ + ("public", "Inh1", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "inh2", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl1", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl2", "table", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl3", "table", POSTGRES_USER, "0 bytes", None), + ] + headers = objects_listing_headers + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dv(executor): + """List all views in public schema.""" + results = executor(r"\dv") + title = None + row = [("public", "vw1", "view", POSTGRES_USER)] + headers = objects_listing_headers[:-2] + status = "SELECT 1" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_dv_verbose(executor): + """List all views in s1 schema in verbose mode.""" + results = executor(r"\dv+ schema1.*") + title = None + row = [("schema1", "s1_vw1", "view", POSTGRES_USER, "0 bytes", None)] + headers = objects_listing_headers + status = "SELECT 1" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_dm(executor): + """List all materialized views in schema1.""" + results = executor(r"\dm schema1.*") + title = None + row = [("schema1", "s1_mvw1", "materialized view", POSTGRES_USER)] + headers = objects_listing_headers[:-2] + status = "SELECT 1" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_dm_verbose(executor): + """List all materialized views in public schema in verbose mode.""" + results = executor(r"\dm+") + title = None + row = [("public", "mvw1", "materialized view", POSTGRES_USER, "8192 bytes", None)] + headers = objects_listing_headers + status = "SELECT 1" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_ds(executor): + """List all sequences in public schema.""" + results = executor(r"\ds") + title = None + row = [("public", "tbl2_id2_seq", "sequence", POSTGRES_USER)] + headers = objects_listing_headers[:-2] + status = "SELECT 1" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_ds_verbose(executor): + """List all sequences in public schema in verbose mode.""" + results = executor(r"\ds+") + title = None + row = [("public", "tbl2_id2_seq", "sequence", POSTGRES_USER, "8192 bytes", None)] + headers = objects_listing_headers + status = "SELECT 1" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_di(executor): + """List all indexes in public schema.""" + results = executor(r"\di") + title = None + row = [ + ("public", "id_text", "index", POSTGRES_USER), + ("public", "tbl3_c3_excl", "index", POSTGRES_USER), + ] + headers = objects_listing_headers[:-2] + status = "SELECT %s" % len(row) + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_di_verbose(executor): + """List all indexes in public schema in verbose mode.""" + results = executor(r"\di+") + title = None + row = [ + ("public", "id_text", "index", POSTGRES_USER, "8192 bytes", None), + ("public", "tbl3_c3_excl", "index", POSTGRES_USER, "8192 bytes", None), + ] + headers = objects_listing_headers + status = "SELECT 2" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_dx(executor): + """List all extensions.""" + results = executor(r"\dx") + title = None + row = [("plpgsql", "1.0", "pg_catalog", "PL/pgSQL procedural language")] + headers = ["Name", "Version", "Schema", "Description"] + status = "SELECT 1" + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_dx_verbose(executor): + """List all extensions in verbose mode.""" + results = executor(r"\dx+") + title = '\nObjects in extension "plpgsql"' + row = [ + ("function plpgsql_call_handler()",), + ("function plpgsql_inline_handler(internal)",), + ("function plpgsql_validator(oid)",), + ("language plpgsql",), + ] + headers = ["Object description"] + status = "SELECT %s" % len(row) + expected = [title, row, headers, status] + assert results == expected + + +@dbtest +def test_slash_dT(executor): + """List all datatypes.""" + results = executor(r"\dT") + title = None + rows = [("public", "foo", None), ("public", "gender_t", None)] + headers = ["Schema", "Name", "Description"] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dD(executor): + title = None + headers = ["Schema", "Name", "Type", "Modifier", "Check"] + results = executor(r"\dD") + rows = [ + ( + "public", + "gender_t", + "character(1)", + "", + "CHECK (VALUE = ANY (ARRAY['F'::bpchar, 'M'::bpchar]))", + ) + ] + status = "SELECT 1" + expected = [title, rows, headers, status] + assert results == expected + + results = executor(r"\dD schema1.*") + rows = [ + ("schema1", "bigint_t", "bigint", "", ""), + ("schema1", "smallint_t", "smallint", "", ""), + ] + status = "SELECT %s" % len(rows) + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dD_verbose(executor): + title = None + headers = [ + "Schema", + "Name", + "Type", + "Modifier", + "Check", + "Access privileges", + "Description", + ] + results = executor(r"\dD+") + rows = [ + ( + "public", + "gender_t", + "character(1)", + "", + "CHECK (VALUE = ANY (ARRAY['F'::bpchar, 'M'::bpchar]))", + None, + None, + ) + ] + status = "SELECT 1" + expected = [title, rows, headers, status] + assert results == expected + + results = executor(r"\dD+ schema1.bigint_t") + rows = [("schema1", "bigint_t", "bigint", "", "", None, "a really large integer")] + status = "SELECT 1" + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_dF(executor): + title, rows, header, status = executor(r"\dF") + assert title is None + assert header == ["Schema", "Name", "Description"] + assert ("pg_catalog", "spanish", "configuration for spanish language") in rows + + results = executor(r"\dD *ian") + assert title is None + assert header == ["Schema", "Name", "Description"] + assert ("pg_catalog", "russian", "configuration for russian language") in rows + + results = executor(r"\dD ge*") + assert title is None + assert header == ["Schema", "Name", "Description"] + assert ("pg_catalog", "german", "configuration for german language") in rows + + +@dbtest +def test_slash_dF_verbose(executor): + results = executor(r"\dF+")[1] + assert ("asciihword", "simple") in results + + results = executor(r"\dF+ *panish")[1] + assert ("asciihword", "spanish_stem") in results + + results = executor(r"\dF+ swed*")[1] + assert ("asciihword", "swedish_stem") in results + + results = executor(r"\dF+ jap") + assert results == [None, None, None, 'Did not find any results for pattern "jap".'] + + +@dbtest +def test_slash_db(executor): + """List all tablespaces.""" + title, rows, header, status = executor(r"\db") + assert title is None + assert header == ["Name", "Owner", "Location"] + assert "pg_default" in rows[0] + + +@dbtest +def test_slash_db_name(executor): + """List tablespace by name.""" + title, rows, header, status = executor(r"\db pg_default") + assert title is None + assert header == ["Name", "Owner", "Location"] + assert "pg_default" in rows[0] + assert status == "SELECT 1" + + +@dbtest +def test_slash_df(executor): + results = executor(r"\df") + title = None + rows = [("public", "func1", "integer", "", "normal")] + headers = ["Schema", "Name", "Result data type", "Argument data types", "Type"] + status = "SELECT 1" + expected = [title, rows, headers, status] + assert results == expected + + +help_rows = [ + [ + "ABORT", + "ALTER AGGREGATE", + "ALTER COLLATION", + "ALTER CONVERSION", + "ALTER DATABASE", + "ALTER DEFAULT PRIVILEGES", + ], + [ + "ALTER DOMAIN", + "ALTER EVENT TRIGGER", + "ALTER EXTENSION", + "ALTER FOREIGN DATA WRAPPER", + "ALTER FOREIGN TABLE", + "ALTER FUNCTION", + ], + [ + "ALTER GROUP", + "ALTER INDEX", + "ALTER LANGUAGE", + "ALTER LARGE OBJECT", + "ALTER MATERIALIZED VIEW", + "ALTER OPCLASS", + ], + [ + "ALTER OPERATOR", + "ALTER OPFAMILY", + "ALTER POLICY", + "ALTER ROLE", + "ALTER RULE", + "ALTER SCHEMA", + ], + [ + "ALTER SEQUENCE", + "ALTER SERVER", + "ALTER SYSTEM", + "ALTER TABLE", + "ALTER TABLESPACE", + "ALTER TRIGGER", + ], + [ + "ALTER TSCONFIG", + "ALTER TSDICTIONARY", + "ALTER TSPARSER", + "ALTER TSTEMPLATE", + "ALTER TYPE", + "ALTER USER", + ], + ["ALTER USER MAPPING", "ALTER VIEW", "ANALYZE", "BEGIN", "CHECKPOINT", "CLOSE"], + ["CLUSTER", "COMMENT", "COMMIT", "COMMIT PREPARED", "COPY", "CREATE AGGREGATE"], + [ + "CREATE CAST", + "CREATE COLLATION", + "CREATE CONVERSION", + "CREATE DATABASE", + "CREATE DOMAIN", + "CREATE EVENT TRIGGER", + ], + [ + "CREATE EXTENSION", + "CREATE FOREIGN DATA WRAPPER", + "CREATE FOREIGN TABLE", + "CREATE FUNCTION", + "CREATE GROUP", + "CREATE INDEX", + ], + [ + "CREATE LANGUAGE", + "CREATE MATERIALIZED VIEW", + "CREATE OPCLASS", + "CREATE OPERATOR", + "CREATE OPFAMILY", + "CREATE POLICY", + ], + [ + "CREATE ROLE", + "CREATE RULE", + "CREATE SCHEMA", + "CREATE SEQUENCE", + "CREATE SERVER", + "CREATE TABLE", + ], + [ + "CREATE TABLE AS", + "CREATE TABLESPACE", + "CREATE TRANSFORM", + "CREATE TRIGGER", + "CREATE TSCONFIG", + "CREATE TSDICTIONARY", + ], + [ + "CREATE TSPARSER", + "CREATE TSTEMPLATE", + "CREATE TYPE", + "CREATE USER", + "CREATE USER MAPPING", + "CREATE VIEW", + ], + ["DEALLOCATE", "DECLARE", "DELETE", "DISCARD", "DO", "DROP AGGREGATE"], + [ + "DROP CAST", + "DROP COLLATION", + "DROP CONVERSION", + "DROP DATABASE", + "DROP DOMAIN", + "DROP EVENT TRIGGER", + ], + [ + "DROP EXTENSION", + "DROP FOREIGN DATA WRAPPER", + "DROP FOREIGN TABLE", + "DROP FUNCTION", + "DROP GROUP", + "DROP INDEX", + ], + [ + "DROP LANGUAGE", + "DROP MATERIALIZED VIEW", + "DROP OPCLASS", + "DROP OPERATOR", + "DROP OPFAMILY", + "DROP OWNED", + ], + [ + "DROP POLICY", + "DROP ROLE", + "DROP RULE", + "DROP SCHEMA", + "DROP SEQUENCE", + "DROP SERVER", + ], + [ + "DROP TABLE", + "DROP TABLESPACE", + "DROP TRANSFORM", + "DROP TRIGGER", + "DROP TSCONFIG", + "DROP TSDICTIONARY", + ], + [ + "DROP TSPARSER", + "DROP TSTEMPLATE", + "DROP TYPE", + "DROP USER", + "DROP USER MAPPING", + "DROP VIEW", + ], + ["END", "EXECUTE", "EXPLAIN", "FETCH", "GRANT", "IMPORT FOREIGN SCHEMA"], + ["INSERT", "LISTEN", "LOAD", "LOCK", "MOVE", "NOTIFY"], + [ + "PGBENCH", + "PREPARE", + "PREPARE TRANSACTION", + "REASSIGN OWNED", + "REFRESH MATERIALIZED VIEW", + "REINDEX", + ], + [ + "RELEASE SAVEPOINT", + "RESET", + "REVOKE", + "ROLLBACK", + "ROLLBACK PREPARED", + "ROLLBACK TO", + ], + ["SAVEPOINT", "SECURITY LABEL", "SELECT", "SELECT INTO", "SET", "SET CONSTRAINTS"], + [ + "SET ROLE", + "SET SESSION AUTH", + "SET TRANSACTION", + "SHOW", + "START TRANSACTION", + "TRUNCATE", + ], + ["UNLISTEN", "UPDATE", "VACUUM", "VALUES"], +] + + +@dbtest +def test_slash_h(executor): + """List all commands.""" + results = executor(r"\h") + expected = [None, help_rows, [], None] + assert results == expected + + +@dbtest +def test_slash_h_command(executor): + """Check help is returned for all commands""" + for command in itertools.chain(*help_rows): + results = executor(r"\h %s" % command) + assert results[3].startswith("Description\n") + assert "Syntax" in results[3] + + +@dbtest +def test_slash_h_alias(executor): + r"""\? is properly aliased to \h""" + h_results = executor(r"\h SELECT") + results = executor(r"\? SELECT") + assert results[3] == h_results[3] + + +@dbtest +def test_slash_copy_to_tsv(executor, tmpdir): + filepath = tmpdir.join("pycons.tsv") + executor( + r"\copy (SELECT 'Montréal', 'Portland', 'Cleveland') TO '{0}' ".format(filepath) + ) + infile = filepath.open(encoding="utf-8") + contents = infile.read() + assert len(contents.splitlines()) == 1 + assert "Montréal" in contents + + +@dbtest +def test_slash_copy_throws_error_without_TO_or_FROM(executor): + with pytest.raises(Exception) as exc_info: + executor("\copy (SELECT 'Montréal', 'Portland', 'Cleveland') INTO stdout ") + assert ( + str(exc_info.value) + == "Missing keyword in \\copy command. Either TO or FROM is required." + ) + + +@dbtest +def test_slash_copy_to_stdout(executor, capsys): + executor(r"\copy (SELECT 'Montréal', 'Portland', 'Cleveland') TO stdout") + (out, err) = capsys.readouterr() + assert out == "Montréal\tPortland\tCleveland\n" + + +@dbtest +def test_slash_copy_to_csv(executor, tmpdir): + filepath = tmpdir.join("pycons.tsv") + executor( + r"\copy (SELECT 'Montréal', 'Portland', 'Cleveland') TO '{0}' WITH csv".format( + filepath + ) + ) + infile = filepath.open(encoding="utf-8") + contents = infile.read() + assert len(contents.splitlines()) == 1 + assert "Montréal" in contents + assert "," in contents + + +@dbtest +def test_slash_copy_from_csv(executor, connection, tmpdir): + filepath = tmpdir.join("tbl1.csv") + executor(r"\copy (SELECT 22, 'elephant') TO '{0}' WITH csv".format(filepath)) + executor(r"\copy tbl1 FROM '{0}' WITH csv".format(filepath)) + cur = connection.cursor() + cur.execute("SELECT * FROM tbl1 WHERE id1 = 22") + row = cur.fetchone() + assert row[1] == "elephant" + + +@dbtest +def test_slash_sf(executor): + results = executor(r"\sf func1") + title = None + rows = [ + ( + "CREATE OR REPLACE FUNCTION public.func1()\n" + " RETURNS integer\n" + " LANGUAGE sql\n" + "AS $function$select 1$function$\n", + ), + ] + headers = ["Source"] + status = None + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_sf_unknown(executor): + try: + executor(r"\sf non_existing") + except Exception as e: + assert "non_existing" in str(e) + else: + assert False, "Expected an exception" + + +@dbtest +def test_slash_sf_parens(executor): + results = executor(r"\sf func1()") + title = None + rows = [ + ( + "CREATE OR REPLACE FUNCTION public.func1()\n" + " RETURNS integer\n" + " LANGUAGE sql\n" + "AS $function$select 1$function$\n", + ), + ] + headers = ["Source"] + status = None + expected = [title, rows, headers, status] + assert results == expected + + +@dbtest +def test_slash_sf_verbose(executor): + results = executor(r"\sf+ schema1.s1_func1") + title = None + rows = [ + ( + " CREATE OR REPLACE FUNCTION schema1.s1_func1()\n" + " RETURNS integer\n" + " LANGUAGE sql\n" + "1 AS $function$select 2$function$\n", + ), + ] + headers = ["Source"] + status = None + expected = [title, rows, headers, status] + assert results == expected + + +@fdw_test +def test_slash_dE(executor): + with foreign_db_environ(): + results = executor(r"\dE") + title = None + rows = [("public", "foreign_foo", "foreign table", "postgres")] + headers = ["Schema", "Name", "Type", "Owner"] + status = "SELECT 1" + expected = [title, rows, headers, status] + assert results == expected + + +@fdw_test +def test_slash_dE_with_pattern(executor): + with foreign_db_environ(): + results = executor(r"\dE foreign_foo") + title = None + rows = [("public", "foreign_foo", "foreign table", "postgres")] + headers = ["Schema", "Name", "Type", "Owner"] + status = "SELECT 1" + expected = [title, rows, headers, status] + assert results == expected + + results = executor(r"\dE *_foo") + assert results == expected + + results = executor(r"\dE no_such_table") + rows = [] + status = "SELECT 0" + expected = [title, rows, headers, status] + assert results == expected + + +@fdw_test +def test_slash_dE_verbose(executor): + with foreign_db_environ(): + results = executor(r"\dE+") + title = None + rows = [("public", "foreign_foo", "foreign table", "postgres", "0 bytes", None)] + headers = ["Schema", "Name", "Type", "Owner", "Size", "Description"] + status = "SELECT 1" + expected = [title, rows, headers, status] + assert results == expected |