test_console.py

#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you 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
#
#    http://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.
import datetime
import os
import uuid
from pathlib import PosixPath
from unittest import mock
from unittest.mock import MagicMock

import pyarrow as pa
import pytest
from click.testing import CliRunner
from pytest_mock import MockFixture

from pyiceberg import __version__
from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.cli.console import run
from pyiceberg.io import WAREHOUSE
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.transforms import IdentityTransform
from pyiceberg.typedef import Properties
from pyiceberg.types import LongType, NestedField
from pyiceberg.utils.config import Config


def test_missing_uri(mocker: MockFixture, empty_home_dir_path: str) -> None:
    # mock to prevent parsing ~/.pyiceberg.yaml or {PYICEBERG_HOME}/.pyiceberg.yaml
    mocker.patch.dict(os.environ, values={"HOME": empty_home_dir_path, "PYICEBERG_HOME": empty_home_dir_path})
    mocker.patch("pyiceberg.catalog._ENV_CONFIG", return_value=Config())

    runner = CliRunner()
    result = runner.invoke(run, ["list"])

    assert result.exit_code == 1
    assert result.output == "Could not initialize catalog with the following properties: {}\n"


def test_hive_catalog_missing_uri_shows_helpful_error(mocker: MockFixture) -> None:
    mock_env_config = mocker.MagicMock()
    mock_env_config.get_catalog_config.return_value = {"type": "hive"}
    mocker.patch("pyiceberg.catalog._ENV_CONFIG", mock_env_config)

    runner = CliRunner()
    result = runner.invoke(run, ["--catalog", "my_hive_catalog", "list"])

    assert result.exit_code == 1
    assert "URI missing, please provide using --uri" in result.output
    assert "'uri'" not in result.output


def test_version_does_not_load_catalog(mocker: MockFixture) -> None:
    mock_load_catalog = mocker.patch("pyiceberg.cli.console.load_catalog", side_effect=Exception("should not be called"))

    runner = CliRunner()
    result = runner.invoke(run, ["version"])

    assert result.exit_code == 0
    assert result.output == f"{__version__}\n"
    mock_load_catalog.assert_not_called()


@pytest.fixture(autouse=True)
def env_vars(mocker: MockFixture) -> None:
    mocker.patch.dict(os.environ, MOCK_ENVIRONMENT)


@pytest.fixture(name="catalog")
def fixture_catalog(mocker: MockFixture, tmp_path: PosixPath) -> InMemoryCatalog:
    in_memory_catalog = InMemoryCatalog(
        "test.in_memory.catalog", **{WAREHOUSE: tmp_path.absolute().as_posix(), "test.key": "test.value"}
    )
    mocker.patch("pyiceberg.cli.console.load_catalog", return_value=in_memory_catalog)
    return in_memory_catalog


@pytest.fixture(name="namespace_properties")
def fixture_namespace_properties() -> Properties:
    return TEST_NAMESPACE_PROPERTIES.copy()


@pytest.fixture()
def mock_datetime_now(monkeypatch: pytest.MonkeyPatch) -> None:
    datetime_mock = MagicMock(wraps=datetime.datetime)
    datetime_mock.now.return_value = datetime.datetime.fromtimestamp(TEST_TIMESTAMP / 1000.0).astimezone()
    monkeypatch.setattr(datetime, "datetime", datetime_mock)


TEST_TABLE_IDENTIFIER = ("default", "my_table")
TEST_TABLE_NAMESPACE = "default"
TEST_NAMESPACE_PROPERTIES = {"location": "s3://warehouse/database/location"}
TEST_TABLE_NAME = "my_table"
TEST_TABLE_SCHEMA = Schema(
    NestedField(1, "x", LongType(), required=True),
    NestedField(2, "y", LongType(), doc="comment", required=True),
    NestedField(3, "z", LongType(), required=True),
)
TEST_TABLE_PARTITION_SPEC = PartitionSpec(PartitionField(name="x", transform=IdentityTransform(), source_id=1, field_id=1000))
TEST_TABLE_PROPERTIES = {"read.split.target.size": "134217728", "write.parquet.bloom-filter-enabled.column.x": True}
TEST_TABLE_UUID = uuid.UUID("d20125c8-7284-442c-9aea-15fee620737c")
TEST_TIMESTAMP = 1602638573874
MOCK_ENVIRONMENT = {"PYICEBERG_CATALOG__PRODUCTION__URI": "test://doesnotexist"}


def test_list_root(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)

    runner = CliRunner()
    result = runner.invoke(run, ["list"])

    assert result.exit_code == 0
    assert TEST_TABLE_NAMESPACE in result.output


def test_list_namespace(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["list", "default"])

    assert result.exit_code == 0
    assert result.output == "default.my_table\n"


def test_describe_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["describe", "default"])

    assert result.exit_code == 0
    assert result.output == "location  s3://warehouse/database/location\n"


def test_describe_namespace_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["describe", "doesnotexist"])

    assert result.exit_code == 1
    assert result.output == "Namespace doesnotexist does not exists\n"


@pytest.fixture()
def test_describe_table(catalog: InMemoryCatalog, mock_datetime_now: None) -> None:
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["describe", "default.my_table"])
    assert result.exit_code == 0
    assert (
        # Strip the whitespace on the end
        "\n".join([line.rstrip() for line in result.output.split("\n")])
        == """Table format version  2
Metadata location     s3://warehouse/default/my_table/metadata/metadata.json
Table UUID            d20125c8-7284-442c-9aea-15fee620737c
Last Updated          1602638573874
Partition spec        [
                        1000: x: identity(1)
                      ]
Sort order            []
Current schema        Schema, id=0
                      ├── 1: x: required long
                      ├── 2: y: required long (comment)
                      └── 3: z: required long
Current snapshot      None
Snapshots             Snapshots
                      └── Snapshot 1925, schema None
Properties
"""
    )


def test_describe_table_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["describe", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Table or namespace does not exist: default.doesnotexist\n"


def test_schema(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["schema", "default.my_table"])
    assert result.exit_code == 0
    assert (
        "\n".join([line.rstrip() for line in result.output.split("\n")])
        == """x  long
y  long  comment
z  long
"""
    )


def test_schema_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["schema", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


def test_spec(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["spec", "default.my_table"])
    assert result.exit_code == 0
    assert (
        result.output
        == """[
  1000: x: identity(1)
]
"""
    )


def test_spec_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["spec", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


@mock.patch("uuid.uuid4", return_value=TEST_TABLE_UUID)
def test_uuid(uuid_mock: None, catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["uuid", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == """d20125c8-7284-442c-9aea-15fee620737c\n"""


def test_uuid_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["uuid", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


def test_location(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )
    runner = CliRunner()
    result = runner.invoke(run, ["location", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == f"""{catalog._warehouse_location}/default/my_table\n"""


def test_location_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["location", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


def test_drop_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["drop", "table", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == """Dropped table: default.my_table\n"""


def test_drop_table_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["drop", "table", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


def test_drop_namespace(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)

    runner = CliRunner()
    result = runner.invoke(run, ["drop", "namespace", "default"])
    assert result.exit_code == 0
    assert result.output == """Dropped namespace: default\n"""


def test_drop_namespace_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["drop", "namespace", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Namespace does not exist: doesnotexist\n"


def test_create_namespace(catalog: InMemoryCatalog) -> None:
    runner = CliRunner()
    result = runner.invoke(run, ["create", "namespace", TEST_TABLE_NAMESPACE])
    assert result.exit_code == 0
    assert result.output == """Created namespace: default\n"""


def test_create_namespace_already_exists(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)

    runner = CliRunner()
    result = runner.invoke(run, ["create", "namespace", TEST_TABLE_NAMESPACE])
    assert result.exit_code == 1
    assert result.output == "Namespace default already exists\n"


def test_rename_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["rename", "default.my_table", "default.my_new_table"])
    assert result.exit_code == 0
    assert result.output == """Renamed table from default.my_table to default.my_new_table\n"""


def test_rename_table_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    catalog.create_namespace("default")
    runner = CliRunner()
    result = runner.invoke(run, ["rename", "default.doesnotexist", "default.bar"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: doesnotexist\n"


def test_properties_get_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "get", "table", "default.my_table"])
    assert result.exit_code == 0
    assert (
        result.output
        == "read.split.target.size                       134217728\nwrite.parquet.bloom-filter-enabled.column.x  true     \n"
    )


def test_properties_get_table_specific_property(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "get", "table", "default.my_table", "read.split.target.size"])
    assert result.exit_code == 0
    assert result.output == "134217728\n"


def test_properties_get_table_specific_property_that_doesnt_exist(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "get", "table", "default.my_table", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Could not find property doesnotexist on table default.my_table\n"


def test_properties_get_table_does_not_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "get", "table", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Empty namespace identifier\n"


def test_properties_get_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "get", "namespace", "default"])
    assert result.exit_code == 0
    assert result.output == "location  s3://warehouse/database/location\n"


def test_properties_get_namespace_specific_property(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "get", "namespace", "default", "location"])
    assert result.exit_code == 0
    assert result.output == "s3://warehouse/database/location\n"


def test_properties_get_namespace_does_not_exist(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "get", "namespace", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Namespace doesnotexist does not exists\n"


def test_properties_set_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "set", "namespace", "default", "location", "s3://new_location"])
    assert result.exit_code == 0
    assert result.output == "Updated location on default\n"


def test_properties_set_namespace_that_doesnt_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "set", "namespace", "doesnotexist", "location", "s3://new_location"])
    assert result.exit_code == 1
    assert result.output == "Namespace doesnotexist does not exists\n"


def test_properties_set_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "set", "table", "default.my_table", "location", "s3://new_location"])
    assert result.exit_code == 0
    assert result.output == "Set location=s3://new_location on default.my_table\n"


def test_properties_set_table_does_not_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "set", "table", "default.doesnotexist", "location", "s3://new_location"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


def test_properties_remove_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "remove", "namespace", "default", "location"])
    assert result.exit_code == 0
    assert result.output == "Property location removed from default\n"


def test_properties_remove_namespace_that_doesnt_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "remove", "namespace", "doesnotexist", "location"])
    assert result.exit_code == 1
    assert result.output == "Namespace doesnotexist does not exists\n"


def test_properties_remove_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "remove", "table", "default.my_table", "read.split.target.size"])
    assert result.exit_code == 0
    assert result.output == "Property read.split.target.size removed from default.my_table\n"


def test_properties_remove_table_property_does_not_exists(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "remove", "table", "default.my_table", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == "Property doesnotexist does not exist on default.my_table\n"


def test_properties_remove_table_does_not_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["properties", "remove", "table", "default.doesnotexist", "location"])
    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


def test_json_list_root(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "list"])
    assert result.exit_code == 0
    assert result.output == """["default"]\n"""


def test_json_list_namespace(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "list", "default"])
    assert result.exit_code == 0
    assert result.output == """["default.my_table"]\n"""


def test_json_describe_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "describe", "default"])
    assert result.exit_code == 0
    assert result.output == """{"location": "s3://warehouse/database/location"}\n"""


def test_json_describe_namespace_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "describe", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchNamespaceError", "message": "Namespace doesnotexist does not exists"}\n"""


@pytest.fixture()
def test_json_describe_table(catalog: InMemoryCatalog, mock_datetime_now: None) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "describe", "default.my_table"])
    assert result.exit_code == 0
    assert (
        result.output
        == """{"identifier":["default","my_table"],"metadata_location":"s3://warehouse/default/my_table/metadata/metadata.json","metadata":{"location":"s3://bucket/test/location","table-uuid":"d20125c8-7284-442c-9aea-15fee620737c","last-updated-ms":1602638573874,"last-column-id":3,"schemas":[{"type":"struct","fields":[{"id":1,"name":"x","type":"long","required":true},{"id":2,"name":"y","type":"long","required":true,"doc":"comment"},{"id":3,"name":"z","type":"long","required":true}],"schema-id":0,"identifier-field-ids":[]}],"current-schema-id":0,"partition-specs":[{"spec-id":0,"fields":[{"source-id":1,"field-id":1000,"transform":"identity","name":"x"}]}],"default-spec-id":0,"last-partition-id":1000,"properties":{},"snapshots":[{"snapshot-id":1925,"timestamp-ms":1602638573822}],"snapshot-log":[],"metadata-log":[],"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0,"refs":{},"format-version":2,"schema":{"type":"struct","fields":[{"id":1,"name":"x","type":"long","required":true},{"id":2,"name":"y","type":"long","required":true,"doc":"comment"},{"id":3,"name":"z","type":"long","required":true}],"schema-id":0,"identifier-field-ids":[]},"partition-spec":[{"source-id":1,"field-id":1000,"transform":"identity","name":"x"}]}}\n"""
    )


def test_json_describe_table_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "describe", "default.doesnotexist"])
    assert result.exit_code == 1
    assert (
        result.output
        == """{"type": "NoSuchTableError", "message": "Table or namespace does not exist: default.doesnotexist"}\n"""
    )


def test_json_schema(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "schema", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == (
        '{"type":"struct","fields":[{"id":1,"name":"x","type":"long","required":true},'
        '{"id":2,"name":"y","type":"long","required":true,"doc":"comment"},'
        '{"id":3,"name":"z","type":"long","required":true}],"schema-id":0,"identifier-field-ids":[]}\n'
    )


def test_json_schema_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "schema", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


def test_json_spec(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "spec", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == """{"spec-id":0,"fields":[{"source-id":1,"field-id":1000,"transform":"identity","name":"x"}]}\n"""


def test_json_spec_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "spec", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


@mock.patch("uuid.uuid4", return_value=TEST_TABLE_UUID)
def test_json_uuid(uuid_mock: None, catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "uuid", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == """{"uuid": "d20125c8-7284-442c-9aea-15fee620737c"}\n"""


def test_json_uuid_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "uuid", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


def test_json_location(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "location", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == f'"{catalog._warehouse_location}/default/my_table"\n'


def test_json_location_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "location", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


def test_json_drop_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "drop", "table", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == """"Dropped table: default.my_table"\n"""


def test_json_drop_table_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "drop", "table", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


def test_json_drop_namespace(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "drop", "namespace", "default"])
    assert result.exit_code == 0
    assert result.output == """"Dropped namespace: default"\n"""


def test_json_drop_namespace_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "drop", "namespace", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchNamespaceError", "message": "Namespace does not exist: doesnotexist"}\n"""


def test_json_rename_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "rename", "default.my_table", "default.my_new_table"])
    assert result.exit_code == 0
    assert result.output == """"Renamed table from default.my_table to default.my_new_table"\n"""


def test_json_rename_table_does_not_exists(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    catalog.create_namespace("default")
    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "rename", "default.doesnotexist", "default.bar"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: doesnotexist"}\n"""


def test_json_properties_get_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "get", "table", "default.my_table"])
    assert result.exit_code == 0
    assert result.output == """{"read.split.target.size": "134217728", "write.parquet.bloom-filter-enabled.column.x": "true"}\n"""


def test_json_properties_get_table_specific_property(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "get", "table", "default.my_table", "read.split.target.size"])
    assert result.exit_code == 0
    assert result.output == """"134217728"\n"""


def test_json_properties_get_table_specific_property_that_doesnt_exist(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "get", "table", "default.my_table", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == (
        '{"type": "NoSuchPropertyException", "message": "Could not find property doesnotexist on table default.my_table"}\n'
    )


def test_json_properties_get_table_does_not_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "get", "table", "default.doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


def test_json_properties_get_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "get", "namespace", "default"])
    assert result.exit_code == 0
    assert result.output == """{"location": "s3://warehouse/database/location"}\n"""


def test_json_properties_get_namespace_specific_property(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "get", "namespace", "default", "location"])
    assert result.exit_code == 0
    assert result.output == """"s3://warehouse/database/location"\n"""


def test_json_properties_get_namespace_does_not_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "get", "namespace", "doesnotexist"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchNamespaceError", "message": "Namespace doesnotexist does not exists"}\n"""


def test_json_properties_set_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "set", "namespace", "default", "location", "s3://new_location"])
    assert result.exit_code == 0
    assert result.output == """"Updated location on default"\n"""


def test_json_properties_set_namespace_that_doesnt_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(
        run, ["--output=json", "properties", "set", "namespace", "doesnotexist", "location", "s3://new_location"]
    )
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchNamespaceError", "message": "Namespace doesnotexist does not exists"}\n"""


def test_json_properties_set_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(
        run, ["--output=json", "properties", "set", "table", "default.my_table", "location", "s3://new_location"]
    )
    assert result.exit_code == 0
    assert result.output == """"Set location=s3://new_location on default.my_table"\n"""


def test_json_properties_set_table_does_not_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(
        run, ["--output=json", "properties", "set", "table", "default.doesnotexist", "location", "s3://new_location"]
    )
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


def test_json_properties_remove_namespace(catalog: InMemoryCatalog, namespace_properties: Properties) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE, namespace_properties)

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "remove", "namespace", "default", "location"])
    assert result.exit_code == 0
    assert result.output == """"Property location removed from default"\n"""


def test_json_properties_remove_namespace_that_doesnt_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "remove", "namespace", "doesnotexist", "location"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchNamespaceError", "message": "Namespace doesnotexist does not exists"}\n"""


def test_json_properties_remove_table(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "remove", "table", "default.my_table", "read.split.target.size"])
    assert result.exit_code == 0
    assert result.output == """"Property read.split.target.size removed from default.my_table"\n"""


def test_json_properties_remove_table_property_does_not_exists(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "remove", "table", "default.my_table", "doesnotexist"])
    assert result.exit_code == 1
    assert (
        result.output
        == """{"type": "NoSuchPropertyException", "message": "Property doesnotexist does not exist on default.my_table"}\n"""
    )


def test_json_properties_remove_table_does_not_exist(catalog: InMemoryCatalog) -> None:
    # pylint: disable=unused-argument

    runner = CliRunner()
    result = runner.invoke(run, ["--output=json", "properties", "remove", "table", "default.doesnotexist", "location"])
    assert result.exit_code == 1
    assert result.output == """{"type": "NoSuchTableError", "message": "Table does not exist: default.doesnotexist"}\n"""


def test_log_level_cli_option(mocker: MockFixture) -> None:
    mock_basicConfig = mocker.patch("logging.basicConfig")

    runner = CliRunner()
    runner.invoke(run, ["--log-level", "DEBUG", "list"])

    # Verify logging.basicConfig was called with DEBUG level
    import logging

    mock_basicConfig.assert_called_once()
    call_kwargs = mock_basicConfig.call_args[1]
    assert call_kwargs["level"] == logging.DEBUG


def test_log_level_env_variable(mocker: MockFixture) -> None:
    mock_basicConfig = mocker.patch("logging.basicConfig")
    mocker.patch.dict(os.environ, {"PYICEBERG_LOG_LEVEL": "INFO"})

    runner = CliRunner()
    runner.invoke(run, ["list"])

    # Verify logging.basicConfig was called with INFO level
    import logging

    mock_basicConfig.assert_called_once()
    call_kwargs = mock_basicConfig.call_args[1]
    assert call_kwargs["level"] == logging.INFO


def test_log_level_default_warning(mocker: MockFixture) -> None:
    mock_basicConfig = mocker.patch("logging.basicConfig")
    # Ensure PYICEBERG_LOG_LEVEL is not set
    mocker.patch.dict(os.environ, {}, clear=False)
    if "PYICEBERG_LOG_LEVEL" in os.environ:
        del os.environ["PYICEBERG_LOG_LEVEL"]

    runner = CliRunner()
    runner.invoke(run, ["list"])

    # Verify logging.basicConfig was called with WARNING level (default)
    import logging

    mock_basicConfig.assert_called_once()
    call_kwargs = mock_basicConfig.call_args[1]
    assert call_kwargs["level"] == logging.WARNING


def test_log_level_cli_overrides_env(mocker: MockFixture) -> None:
    mock_basicConfig = mocker.patch("logging.basicConfig")
    mocker.patch.dict(os.environ, {"PYICEBERG_LOG_LEVEL": "INFO"})

    runner = CliRunner()
    runner.invoke(run, ["--log-level", "ERROR", "list"])

    # Verify CLI option overrides environment variable
    import logging

    mock_basicConfig.assert_called_once()
    call_kwargs = mock_basicConfig.call_args[1]
    assert call_kwargs["level"] == logging.ERROR


def test_warehouse_cli_option_forwarded_to_catalog(mocker: MockFixture) -> None:
    mock_basicConfig = mocker.patch("logging.basicConfig")
    mock_catalog = MagicMock(spec=InMemoryCatalog)
    mock_catalog.list_tables.return_value = []
    mock_catalog.list_namespaces.return_value = []
    mock_load_catalog = mocker.patch("pyiceberg.cli.console.load_catalog", return_value=mock_catalog)

    runner = CliRunner()
    result = runner.invoke(
        run, ["--catalog", "rest", "--uri", "https://catalog.service", "--warehouse", "example-warehouse", "list"]
    )

    assert result.exit_code == 0
    mock_basicConfig.assert_called_once()
    mock_load_catalog.assert_called_once_with("rest", uri="https://catalog.service", warehouse="example-warehouse")


def _create_table_with_expirable_snapshot(catalog: InMemoryCatalog) -> int:
    """Create a table with two snapshots and return the older (non-HEAD) one.

    The HEAD snapshot of a branch is protected from expiration, so to test the
    expire-snapshots command we need a snapshot that has been superseded.
    """
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    table = catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )
    arrow_schema = pa.schema(
        [
            pa.field("x", pa.int64(), nullable=False),
            pa.field("y", pa.int64(), nullable=False),
            pa.field("z", pa.int64(), nullable=False),
        ]
    )
    table.append(pa.Table.from_pylist([{"x": 1, "y": 2, "z": 3}], schema=arrow_schema))
    table.refresh()
    older_snapshot = table.current_snapshot()
    assert older_snapshot is not None
    older_snapshot_id = older_snapshot.snapshot_id
    table.append(pa.Table.from_pylist([{"x": 4, "y": 5, "z": 6}], schema=arrow_schema))
    return older_snapshot_id


def test_expire_snapshots_requires_option(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["expire-snapshots", "default.my_table"])

    assert result.exit_code == 1
    assert "Must provide at least one of --snapshot-id or --older-than." in result.output


def test_expire_snapshots_table_does_not_exists(catalog: InMemoryCatalog) -> None:
    runner = CliRunner()
    result = runner.invoke(run, ["expire-snapshots", "default.doesnotexist", "--snapshot-id", "1"])

    assert result.exit_code == 1
    assert result.output == "Table does not exist: default.doesnotexist\n"


def test_expire_snapshots_by_id(catalog: InMemoryCatalog) -> None:
    snapshot_id = _create_table_with_expirable_snapshot(catalog)

    runner = CliRunner()
    result = runner.invoke(run, ["expire-snapshots", "default.my_table", "--snapshot-id", str(snapshot_id)])

    assert result.exit_code == 0
    assert result.output == "Expired snapshots on default.my_table\n"

    refreshed = catalog.load_table(TEST_TABLE_IDENTIFIER)
    assert refreshed.metadata.snapshot_by_id(snapshot_id) is None


def test_expire_snapshots_older_than(catalog: InMemoryCatalog) -> None:
    snapshot_id = _create_table_with_expirable_snapshot(catalog)
    cutoff = datetime.datetime.now() + datetime.timedelta(days=1)

    runner = CliRunner()
    result = runner.invoke(
        run,
        [
            "expire-snapshots",
            "default.my_table",
            "--older-than",
            cutoff.strftime("%Y-%m-%dT%H:%M:%S"),
        ],
    )

    assert result.exit_code == 0
    assert result.output == "Expired snapshots on default.my_table\n"

    refreshed = catalog.load_table(TEST_TABLE_IDENTIFIER)
    assert refreshed.metadata.snapshot_by_id(snapshot_id) is None


def test_expire_snapshots_unknown_snapshot_id(catalog: InMemoryCatalog) -> None:
    catalog.create_namespace(TEST_TABLE_NAMESPACE)
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
    )

    runner = CliRunner()
    result = runner.invoke(run, ["expire-snapshots", "default.my_table", "--snapshot-id", "999"])

    assert result.exit_code == 1
    assert "Snapshot with ID 999 does not exist." in result.output


def test_json_expire_snapshots_by_id(catalog: InMemoryCatalog) -> None:
    snapshot_id = _create_table_with_expirable_snapshot(catalog)

    runner = CliRunner()
    result = runner.invoke(
        run,
        ["--output=json", "expire-snapshots", "default.my_table", "--snapshot-id", str(snapshot_id)],
    )

    assert result.exit_code == 0
    assert result.output == '"Expired snapshots on default.my_table"\n'