#!/usr/bin/env pytest
###############################################################################
# $Id$
#
# Project:  GDAL/OGR Test Suite
# Purpose:  Test /vsiadls
# Author:   Even Rouault <even dot rouault at spatialys dot com>
#
###############################################################################
# Copyright (c) 2020 Even Rouault <even dot rouault at spatialys dot com>
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the "Software"),
# to deal in the Software without restriction, including without limitation
# the rights to use, copy, modify, merge, publish, distribute, sublicense,
# and/or sell copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
###############################################################################

import base64
import stat

import pytest

from osgeo import gdal

pytestmark = pytest.mark.require_curl()


def open_for_read(uri):
    """
    Opens a test file for reading.
    """
    return gdal.VSIFOpenExL(uri, "rb", 1)


###############################################################################
# Nominal cases (require valid credentials)


def test_vsiadls_real_instance_tests():

    adls_resource = gdal.GetConfigOption("ADLS_RESOURCE")
    if adls_resource is None:
        pytest.skip("Missing ADLS_RESOURCE")

    if "/" not in adls_resource:
        path = "/vsiadls/" + adls_resource

        try:
            statres = gdal.VSIStatL(path)
            assert statres is not None and stat.S_ISDIR(statres.mode), (
                "%s is not a valid bucket" % path
            )

            readdir = gdal.ReadDir(path)
            assert readdir is not None, "ReadDir() should not return empty list"
            for filename in readdir:
                if filename != ".":
                    subpath = path + "/" + filename
                    assert gdal.VSIStatL(subpath) is not None, (
                        "Stat(%s) should not return an error" % subpath
                    )

            unique_id = "vsiadls_test"
            subpath = path + "/" + unique_id
            ret = gdal.Mkdir(subpath, 0)
            assert ret >= 0, "Mkdir(%s) should not return an error" % subpath

            readdir = gdal.ReadDir(path)
            assert unique_id in readdir, "ReadDir(%s) should contain %s" % (
                path,
                unique_id,
            )

            ret = gdal.Mkdir(subpath, 0)
            assert ret != 0, "Mkdir(%s) repeated should return an error" % subpath

            ret = gdal.Rmdir(subpath)
            assert ret >= 0, "Rmdir(%s) should not return an error" % subpath

            readdir = gdal.ReadDir(path)
            assert unique_id not in readdir, "ReadDir(%s) should not contain %s" % (
                path,
                unique_id,
            )

            ret = gdal.Rmdir(subpath)
            assert ret != 0, "Rmdir(%s) repeated should return an error" % subpath

            ret = gdal.Mkdir(subpath, 0)
            assert ret >= 0, "Mkdir(%s) should not return an error" % subpath

            f = gdal.VSIFOpenL(subpath + "/test.txt", "wb")
            assert f is not None
            gdal.VSIFWriteL("hello", 1, 5, f)
            gdal.VSIFCloseL(f)

            ret = gdal.Rmdir(subpath)
            assert ret != 0, (
                "Rmdir(%s) on non empty directory should return an error" % subpath
            )

            f = gdal.VSIFOpenL(subpath + "/test.txt", "rb")
            assert f is not None
            data = gdal.VSIFReadL(1, 5, f).decode("utf-8")
            assert data == "hello"
            gdal.VSIFCloseL(f)

            assert gdal.VSIStatL(subpath + "/test.txt") is not None

            md = gdal.GetFileMetadata(subpath + "/test.txt", "HEADERS")
            assert "x-ms-properties" in md

            md = gdal.GetFileMetadata(subpath + "/test.txt", "STATUS")
            assert "x-ms-resource-type" in md
            assert "x-ms-properties" not in md

            md = gdal.GetFileMetadata(subpath + "/test.txt", "ACL")
            assert "x-ms-acl" in md
            assert "x-ms-permissions" in md

            # Change properties
            properties_foo_bar = "foo=" + base64.b64encode("bar")
            assert gdal.SetFileMetadata(
                subpath + "/test.txt",
                {"x-ms-properties": properties_foo_bar},
                "PROPERTIES",
            )

            md = gdal.GetFileMetadata(subpath + "/test.txt", "HEADERS")
            assert "x-ms-properties" in md
            assert md["x-ms-properties"] == properties_foo_bar

            # Change ACL
            assert gdal.SetFileMetadata(
                subpath + "/test.txt", {"x-ms-permissions": "0777"}, "ACL"
            )

            md = gdal.GetFileMetadata(subpath + "/test.txt", "ACL")
            assert "x-ms-permissions" in md
            assert md["x-ms-permissions"] == "rwxrwxrwx"

            # Change ACL recursively
            md = gdal.GetFileMetadata(subpath, "ACL")
            assert "x-ms-acl" in md
            assert gdal.SetFileMetadata(
                subpath + "/test.txt",
                {"x-ms-acl": md["x-ms-acl"]},
                "ACL",
                ["RECURSIVE=YES", "MODE=set"],
            )

            assert gdal.Rename(subpath + "/test.txt", subpath + "/test2.txt") == 0

            assert gdal.VSIStatL(subpath + "/test.txt") is None

            assert gdal.VSIStatL(subpath + "/test2.txt") is not None

            f = gdal.VSIFOpenL(subpath + "/test2.txt", "rb")
            assert f is not None
            data = gdal.VSIFReadL(1, 5, f).decode("utf-8")
            assert data == "hello"
            gdal.VSIFCloseL(f)

            ret = gdal.Unlink(subpath + "/test2.txt")
            assert ret >= 0, "Unlink(%s) should not return an error" % (
                subpath + "/test2.txt"
            )

            assert gdal.VSIStatL(subpath + "/test2.txt") is None

            assert (
                gdal.Unlink(subpath + "/test2.txt") != 0
            ), "Unlink on a deleted file should return an error"

            f = gdal.VSIFOpenL(subpath + "/test2.txt", "wb")
            assert f is not None
            gdal.VSIFCloseL(f)

            assert gdal.VSIStatL(subpath + "/test2.txt") is not None

        finally:
            assert gdal.RmdirRecursive(subpath) == 0

        return

    f = open_for_read("/vsiadls/" + adls_resource)
    assert f is not None
    ret = gdal.VSIFReadL(1, 1, f)
    gdal.VSIFCloseL(f)

    assert len(ret) == 1

    # Test GetSignedURL()
    signed_url = gdal.GetSignedURL("/vsiadls/" + adls_resource)
    f = open_for_read("/vsicurl_streaming/" + signed_url)
    assert f is not None
    ret = gdal.VSIFReadL(1, 1, f)
    gdal.VSIFCloseL(f)

    assert len(ret) == 1


###############################################################################
# Nominal cases (require valid credentials)
# Note: that test must be run with a delay > 30 seconds due to such a delay
# for re-creating a filesystem of the same name of one that has been destroyed


def test_vsiadls_real_instance_filesystem_tests():

    if gdal.GetConfigOption("ADLS_ALLOW_FILESYSTEM_TESTS") is None:
        pytest.skip("Missing ADLS_ALLOW_FILESYSTEM_TESTS")

    fspath = "/vsiadls/test-vsiadls-filesystem-tests"

    try:
        assert gdal.VSIStatL(fspath) is None

        assert gdal.Mkdir(fspath, 0) == 0

        statres = gdal.VSIStatL(fspath)
        assert statres is not None and stat.S_ISDIR(statres.mode)

        assert gdal.ReadDir(fspath) == ["."]

        assert gdal.Mkdir(fspath, 0) != 0

        assert gdal.Mkdir(fspath + "/subdir", 0) == 0

        statres = gdal.VSIStatL(fspath + "/subdir")
        assert statres is not None and stat.S_ISDIR(statres.mode)

        assert gdal.Rmdir(fspath) != 0

    finally:
        assert gdal.RmdirRecursive(fspath) == 0

        assert gdal.VSIStatL(fspath) is None