#------------------------------------------------------------------------------
# Copyright (c) 2020, 2022, Oracle and/or its affiliates.
#
# This software is dual-licensed to you under the Universal Permissive License
# (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
# 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
# either license.
#
# If you elect to accept the software under the Apache License, Version 2.0,
# the following applies:
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#------------------------------------------------------------------------------

"""
1900 - Module for testing LOB (CLOB and BLOB) variables
"""

import pickle

import oracledb
import test_env

class TestCase(test_env.BaseTestCase):

    def __get_temp_lobs(self, sid):
        cursor = self.connection.cursor()
        cursor.execute("""
                select cache_lobs + nocache_lobs + abstract_lobs
                from v$temporary_lobs
                where sid = :sid""", sid = sid)
        row = cursor.fetchone()
        if row is None:
            return 0
        return int(row[0])

    def __perform_test(self, lob_type, input_type):
        long_string = ""
        db_type = getattr(oracledb, "DB_TYPE_" + lob_type)
        self.cursor.execute(f"truncate table Test{lob_type}s")
        for i in range(0, 11):
            if i > 0:
                char = chr(ord('A') + i - 1)
                long_string += char * 25000
            elif input_type is not db_type:
                continue
            self.cursor.setinputsizes(long_string=input_type)
            if lob_type == "BLOB":
                bind_value = long_string.encode()
            else:
                bind_value = long_string
            self.cursor.execute(f"""
                    insert into Test{lob_type}s (
                        IntCol,
                        {lob_type}Col
                    ) values (
                        :integer_value,
                        :long_string
                    )""",
                    integer_value=i,
                    long_string=bind_value)
        self.connection.commit()
        self.cursor.execute(f"""
                select
                    IntCol,
                    {lob_type}Col
                from Test{lob_type}s
                order by IntCol""")
        self.__validate_query(self.cursor, lob_type)

    def __test_bind_ordering(self, lob_type):
        main_col = "A" * 32768
        extra_col_1 = "B" * 65536
        extra_col_2 = "C" * 131072
        if lob_type == "BLOB":
            main_col = main_col.encode()
            extra_col_1 = extra_col_1.encode()
            extra_col_2 = extra_col_2.encode()
        self.connection.stmtcachesize = 0
        self.cursor.execute(f"truncate table Test{lob_type}s")
        data = (1, main_col, 8, extra_col_1, 15, extra_col_2)
        self.cursor.execute(f"""
                insert into Test{lob_type}s (
                    IntCol,
                    {lob_type}Col,
                    ExtraNumCol1,
                    Extra{lob_type}Col1,
                    ExtraNumCol2,
                    Extra{lob_type}Col2
                ) values (:1, :2, :3, :4, :5, :6)""",
                data)
        with test_env.FetchLobsContextManager(False):
            self.cursor.execute(f"select * from Test{lob_type}s")
            fetched_data = self.cursor.fetchone()
            self.assertEqual(fetched_data, data)

    def __test_fetch_lobs_direct(self, lob_type):
        self.cursor.execute(f"truncate table Test{lob_type}s")
        data = []
        long_string = ""
        for i in range(1, 11):
            if i > 0:
                char = chr(ord('A') + i - 1)
                long_string += char * 25000
            if lob_type == "BLOB":
                data.append((i, long_string.encode()))
            else:
                data.append((i, long_string))
        self.cursor.executemany(f"""
                    insert into Test{lob_type}s (
                        IntCol,
                        {lob_type}Col
                    ) values (
                        :1,
                        :2
                    )""", data)
        with test_env.FetchLobsContextManager(False):
            self.cursor.execute(f"""
                    select
                        IntCol,
                        {lob_type}Col
                    from Test{lob_type}s
                    order by IntCol""")
            self.assertEqual(data, self.cursor.fetchall())

    def __test_lob_operations(self, lob_type):
        self.cursor.execute(f"truncate table Test{lob_type}s")
        self.cursor.setinputsizes(long_string=getattr(oracledb, lob_type))
        long_string = "X" * 75000
        write_value = "TEST"
        if lob_type == "BLOB":
            long_string = long_string.encode("ascii")
            write_value = write_value.encode("ascii")
        self.cursor.execute(f"""
                insert into Test{lob_type}s (
                    IntCol,
                    {lob_type}Col
                ) values (
                    :integer_value,
                    :long_string
                )""",
                integer_value=1,
                long_string=long_string)
        self.cursor.execute(f"""
                select {lob_type}Col
                from Test{lob_type}s
                where IntCol = 1""")
        lob, = self.cursor.fetchone()
        self.assertEqual(lob.isopen(), False)
        lob.open()
        self.assertEqual(lob.isopen(), True)
        lob.close()
        self.assertEqual(lob.isopen(), False)
        self.assertEqual(lob.size(), 75000)
        lob.write(write_value, 75001)
        self.assertEqual(lob.size(), 75000 + len(write_value))
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-2030:",
                               lob.read, 0)
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-2030:",
                               lob.read, -25)
        self.assertEqual(lob.read(), long_string + write_value)
        lob.write(write_value, 1)
        self.assertEqual(lob.read(),
                         write_value + long_string[4:] + write_value)
        lob.trim(25000)
        self.assertEqual(lob.size(), 25000)
        lob.trim(newSize=10000)
        self.assertEqual(lob.size(), 10000)
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-2014:",
                               lob.trim, new_size=50, newSize=60)
        lob.trim()
        self.assertEqual(lob.size(), 0)
        self.assertIsInstance(lob.getchunksize(), int)

    def __test_pickle(self, lob_type):
        value = "A test string value for pickling"
        if lob_type == "BLOB":
            value = value.encode("ascii")
        db_type = getattr(oracledb, "DB_TYPE_" + lob_type)
        lob = self.connection.createlob(db_type)
        lob.write(value)
        pickled_data = pickle.dumps(lob)
        unpickled_value = pickle.loads(pickled_data)
        self.assertEqual(unpickled_value, value)

    def __test_temporary_lob(self, lob_type):
        self.cursor.execute(f"truncate table Test{lob_type}s")
        value = "A test string value"
        if lob_type == "BLOB":
            value = value.encode("ascii")
        db_type = getattr(oracledb, "DB_TYPE_" + lob_type)
        lob = self.connection.createlob(db_type)
        lob.write(value)
        self.cursor.execute("""
                insert into Test%ss (IntCol, %sCol)
                values (:int_val, :lob_val)""" % (lob_type, lob_type),
                int_val=1,
                lob_val=lob)
        self.cursor.execute("select %sCol from Test%ss" % (lob_type, lob_type))
        lob, = self.cursor.fetchone()
        self.assertEqual(lob.read(), value)

    def __validate_query(self, rows, lob_type):
        long_string = ""
        db_type = getattr(oracledb, "DB_TYPE_" + lob_type)
        for row in rows:
            integer_value, lob = row
            self.assertEqual(lob.type, db_type)
            if integer_value == 0:
                self.assertEqual(lob.size(), 0)
                expected_value = ""
                if lob_type == "BLOB":
                    expected_value = expected_value.encode()
                self.assertEqual(lob.read(), expected_value)
            else:
                char = chr(ord('A') + integer_value - 1)
                prev_char = chr(ord('A') + integer_value - 2)
                long_string += char * 25000
                if lob_type == "BLOB":
                    expected_value = long_string.encode("ascii")
                    char = char.encode("ascii")
                    prev_char = prev_char.encode("ascii")
                else:
                    expected_value = long_string
                self.assertEqual(lob.size(), len(expected_value))
                self.assertEqual(lob.read(), expected_value)
                if lob_type == "CLOB":
                    self.assertEqual(str(lob), expected_value)
                self.assertEqual(lob.read(len(expected_value)), char)
            if integer_value > 1:
                offset = (integer_value - 1) * 25000 - 4
                string = prev_char * 5 + char * 5
                self.assertEqual(lob.read(offset, 10), string)

    def test_1900_bind_lob_value(self):
        "1900 - test binding a LOB value directly"
        self.cursor.execute("truncate table TestCLOBs")
        self.cursor.execute("""
                insert into TestCLOBs
                (IntCol, ClobCol)
                values (1, 'Short value')""")
        self.cursor.execute("select ClobCol from TestCLOBs")
        lob, = self.cursor.fetchone()
        self.cursor.execute("""
                insert into TestCLOBs
                (IntCol, ClobCol)
                values (2, :value)""",
                value=lob)

    def test_1901_blob_cursor_description(self):
        "1901 - test cursor description is accurate for BLOBs"
        self.cursor.execute("select IntCol, BlobCol from TestBLOBs")
        expected_value = [
            ('INTCOL', oracledb.DB_TYPE_NUMBER, 10, None, 9, 0, 0),
            ('BLOBCOL', oracledb.DB_TYPE_BLOB, None, None, None, None, 0)
        ]
        self.assertEqual(self.cursor.description, expected_value)

    def test_1902_blob_direct(self):
        "1902 - test binding and fetching BLOB data (directly)"
        self.__perform_test("BLOB", oracledb.DB_TYPE_BLOB)

    def test_1903_blob_indirect(self):
        "1903 - test binding and fetching BLOB data (indirectly)"
        self.__perform_test("BLOB", oracledb.DB_TYPE_LONG_RAW)

    def test_1904_blob_operations(self):
        "1904 - test operations on BLOBs"
        self.__test_lob_operations("BLOB")

    def test_1905_clob_cursor_description(self):
        "1905 - test cursor description is accurate for CLOBs"
        self.cursor.execute("select IntCol, ClobCol from TestCLOBs")
        expected_value = [
            ('INTCOL', oracledb.DB_TYPE_NUMBER, 10, None, 9, 0, False),
            ('CLOBCOL', oracledb.DB_TYPE_CLOB, None, None, None, None, False)
        ]
        self.assertEqual(self.cursor.description, expected_value)

    def test_1906_clob_direct(self):
        "1906 - test binding and fetching CLOB data (directly)"
        self.__perform_test("CLOB", oracledb.DB_TYPE_CLOB)

    def test_1907_clob_indirect(self):
        "1907 - test binding and fetching CLOB data (indirectly)"
        self.__perform_test("CLOB", oracledb.DB_TYPE_LONG)

    def test_1908_clob_operations(self):
        "1908 - test operations on CLOBs"
        self.__test_lob_operations("CLOB")

    def test_1909_create_temp_blob(self):
        "1909 - test creating a temporary BLOB"
        self.__test_temporary_lob("BLOB")

    def test_1910_create_temp_clob(self):
        "1910 - test creating a temporary CLOB"
        self.__test_temporary_lob("CLOB")

    def test_1911_create_temp_nclob(self):
        "1911 - test creating a temporary NCLOB"
        self.__test_temporary_lob("NCLOB")

    def test_1912_multiple_fetch(self):
        "1912 - test retrieving data from a CLOB after multiple fetches"
        self.cursor.arraysize = 1
        self.cursor.execute("select * from TestCLOBS")
        rows = self.cursor.fetchall()
        self.__validate_query(rows, "CLOB")

    def test_1913_nclob_cursor_description(self):
        "1913 - test cursor description is accurate for NCLOBs"
        self.cursor.execute("select IntCol, NClobCol from TestNCLOBs")
        expected_value = [
            ('INTCOL', oracledb.DB_TYPE_NUMBER, 10, None, 9, 0, 0),
            ('NCLOBCOL', oracledb.DB_TYPE_NCLOB, None, None, None, None, 0)
        ]
        self.assertEqual(self.cursor.description, expected_value)

    def test_1914_nclob_direct(self):
        "1914 - test binding and fetching NCLOB data (directly)"
        self.__perform_test("NCLOB", oracledb.DB_TYPE_NCLOB)

    def test_1915_nclob_non_ascii_chars(self):
        "1915 - test binding and fetching NCLOB data (with non-ASCII chars)"
        value = "\u03b4\u4e2a"
        self.cursor.execute("truncate table TestNCLOBs")
        self.cursor.setinputsizes(val=oracledb.DB_TYPE_NVARCHAR)
        self.cursor.execute("""
                insert into TestNCLOBs (IntCol, NClobCol)
                values (1, :val)""",
                val=value)
        self.cursor.execute("select NCLOBCol from TestNCLOBs")
        nclob, = self.cursor.fetchone()
        self.cursor.setinputsizes(val=oracledb.DB_TYPE_NVARCHAR)
        self.cursor.execute("update TestNCLOBs set NCLOBCol = :val",
                            val=nclob.read() + value)
        self.cursor.execute("select NCLOBCol from TestNCLOBs")
        nclob, = self.cursor.fetchone()
        self.assertEqual(nclob.read(), value + value)

    def test_1916_nclob_indirect(self):
        "1916 - test binding and fetching NCLOB data (indirectly)"
        self.__perform_test("NCLOB", oracledb.DB_TYPE_LONG)

    def test_1917_nclob_operations(self):
        "1917 - test operations on NCLOBs"
        self.__test_lob_operations("NCLOB")

    def test_1918_temporary_lobs(self):
        "1918 - test temporary LOBs"
        self.cursor.execute("""
                select sys_context('USERENV', 'SID')
                from dual""")
        sid, = self.cursor.fetchone()
        temp_lobs = self.__get_temp_lobs(sid)
        with self.connection.cursor() as cursor:
            cursor.arraysize = 27
            self.assertEqual(temp_lobs, 0)
            cursor.execute("""
                    select extract(xmlcol, '/').getclobval()
                    from TestXML""")
            for lob, in cursor:
                value = lob.read()
                del lob
        temp_lobs = self.__get_temp_lobs(sid)
        self.assertEqual(temp_lobs, 0)

    def test_1919_assign_string_beyond_array_size(self):
        "1919 - test assign string to NCLOB beyond array size"
        nclobVar = self.cursor.var(oracledb.DB_TYPE_NCLOB)
        self.assertRaises(IndexError, nclobVar.setvalue, 1, "test char")

    def test_1920_supplemental_characters(self):
        "1920 - test read/write temporary LOBs using supplemental characters"
        self.cursor.execute("""
                select value
                from nls_database_parameters
                where parameter = 'NLS_CHARACTERSET'""")
        charset, = self.cursor.fetchone()
        if charset != "AL32UTF8":
            self.skipTest("Database character set must be AL32UTF8")
        supplemental_chars = "𠜎 𠜱 𠝹 𠱓 𠱸 𠲖 𠳏 𠳕 𠴕 𠵼 𠵿 𠸎 𠸏 𠹷 𠺝 " \
                "𠺢 𠻗 𠻹 𠻺 𠼭 𠼮 𠽌 𠾴 𠾼 𠿪 𡁜 𡁯 𡁵 𡁶 𡁻 𡃁 𡃉 𡇙 𢃇 " \
                "𢞵 𢫕 𢭃 𢯊 𢱑 𢱕 𢳂 𢴈 𢵌 𢵧 𢺳 𣲷 𤓓 𤶸 𤷪 𥄫 𦉘 𦟌 𦧲 " \
                "𦧺 𧨾 𨅝 𨈇 𨋢 𨳊 𨳍 𨳒 𩶘"
        self.cursor.execute("truncate table TestCLOBs")
        lob = self.connection.createlob(oracledb.DB_TYPE_CLOB)
        lob.write(supplemental_chars)
        self.cursor.execute("""
                insert into TestCLOBs
                (IntCol, ClobCol)
                values (1, :val)""",
                [lob])
        self.connection.commit()
        self.cursor.execute("select ClobCol from TestCLOBs")
        lob, = self.cursor.fetchone()
        self.assertEqual(lob.read(), supplemental_chars)

    def test_1921_plsql_auto_conversion_clob(self):
        "1921 - test automatic conversion to CLOB for PL/SQL"
        var = self.cursor.var(str, outconverter=lambda v: v[-15:])
        var.setvalue(0, "A" * 50000)
        self.cursor.execute("""
                declare
                    t_Clob          clob;
                begin
                    t_Clob := :data;
                    dbms_lob.copy(:data, t_Clob, 50000);
                    dbms_lob.writeappend(:data, 5, 'BBBBB');
                end;""",
                data=var)
        result = var.getvalue()
        self.assertEqual(result, "A" * 10 + "B" * 5)

    def test_1922_plsql_auto_conversion_nclob(self):
        "1922 - test automatic conversion to NCLOB for PL/SQL"
        var = self.cursor.var(oracledb.DB_TYPE_NCHAR,
                              outconverter=lambda v: v[-12:])
        var.setvalue(0, "N" * 51234)
        self.cursor.execute("""
                declare
                    t_Clob          nclob;
                begin
                    t_Clob := :data;
                    dbms_lob.copy(:data, t_Clob, 51234);
                    dbms_lob.writeappend(:data, 7, 'PPPPPPP');
                end;""",
                data=var)
        result = var.getvalue()
        self.assertEqual(result, "N" * 5 + "P" * 7)

    def test_1923_plsql_auto_conversion_blob(self):
        "1923 - test automatic conversion to BLOB for PL/SQL"
        var = self.cursor.var(bytes,
                              outconverter=lambda v: v[-14:])
        var.setvalue(0, b"L" * 52345)
        self.cursor.execute("""
                declare
                    t_Blob          blob;
                begin
                    t_Blob := :data;
                    dbms_lob.copy(:data, t_Blob, 52345);
                    dbms_lob.writeappend(:data, 6, '515151515151');
                end;""",
                data=var)
        result = var.getvalue()
        self.assertEqual(result, b"L" * 8 + b"Q" * 6)

    def test_1924_pickle_blob(self):
        "1924 - test pickling of BLOB"
        self.__test_pickle("BLOB")

    def test_1925_pickle_clob(self):
        "1925 - test pickling of CLOB"
        self.__test_pickle("CLOB")

    def test_1926_pickle_nclob(self):
        "1925 - test pickling of NCLOB"
        self.__test_pickle("NCLOB")

    def test_1927_fetch_blob_as_bytes(self):
        "1927 - test fetching BLOB as bytes"
        self.__test_fetch_lobs_direct("BLOB")

    def test_1928_fetch_clob_as_str(self):
        "1928 - test fetching CLOB as str"
        self.__test_fetch_lobs_direct("CLOB")

    def test_1929_fetch_nclob_as_str(self):
        "1929 - test fetching NCLOB as str"
        self.__test_fetch_lobs_direct("NCLOB")

    def test_1930_bind_order_blob(self):
        "1930 - test bind ordering with BLOB"
        self.__test_bind_ordering("BLOB")

    def test_1931_bind_order_clob(self):
        "1931 - test bind ordering with CLOB"
        self.__test_bind_ordering("CLOB")

    def test_1932_bind_order_nclob(self):
        "1932 - test bind ordering with NCLOB"
        self.__test_bind_ordering("NCLOB")

if __name__ == "__main__":
    test_env.run_test_cases()