# Copyright (C) 2010-2024 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import io
from base64 import b64encode

import pytest
from unittest.mock import patch, Mock, MagicMock

from rhodecode.lib.middleware.simplesvn import SimpleSvn, SimpleSvnApp
from rhodecode.lib.utils import get_rhodecode_repo_store_path
from rhodecode.tests import SVN_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS


class TestSimpleSvn(object):
    @pytest.fixture(autouse=True)
    def simple_svn(self, baseapp, request_stub):
        base_path = get_rhodecode_repo_store_path()
        self.app = SimpleSvn(config={"auth_ret_code": "", "repo_store.path": base_path}, registry=request_stub.registry)

    def test_get_config(self):
        extras = {"foo": "FOO", "bar": "BAR"}
        config = self.app._create_config(extras, repo_name="test-repo")
        assert config == extras

    @pytest.mark.parametrize("method", ["OPTIONS", "PROPFIND", "GET", "REPORT"])
    def test_get_action_returns_pull(self, method):
        environment = {"REQUEST_METHOD": method}
        action = self.app._get_action(environment)
        assert action == "pull"

    @pytest.mark.parametrize(
        "method",
        ["MKACTIVITY", "PROPPATCH", "PUT", "CHECKOUT", "MKCOL", "MOVE", "COPY", "DELETE", "LOCK", "UNLOCK", "MERGE"],
    )
    def test_get_action_returns_push(self, method):
        environment = {"REQUEST_METHOD": method}
        action = self.app._get_action(environment)
        assert action == "push"

    @pytest.mark.parametrize(
        "path, expected_name",
        [
            ("/hello-svn", "hello-svn"),
            ("/hello-svn/", "hello-svn"),
            ("/group/hello-svn/", "group/hello-svn"),
            ("/group/hello-svn/!svn/vcc/default", "group/hello-svn"),
        ],
    )
    def test_get_repository_name(self, path, expected_name):
        environment = {"PATH_INFO": path}
        name = self.app._get_repository_name(environment)
        assert name == expected_name

    def test_get_repository_name_subfolder(self, backend_svn):
        repo = backend_svn.repo
        environment = {"PATH_INFO": "/{}/path/with/subfolders".format(repo.repo_name)}
        name = self.app._get_repository_name(environment)
        assert name == repo.repo_name

    def test_create_wsgi_app(self):
        with patch.object(SimpleSvn, "_is_svn_enabled") as mock_method:
            mock_method.return_value = False
            with patch("rhodecode.lib.middleware.simplesvn.DisabledSimpleSvnApp") as (wsgi_app_mock):
                config = Mock()
                wsgi_app = self.app._create_wsgi_app(repo_path="", repo_name="", config=config)

            wsgi_app_mock.assert_called_once_with(config)
            assert wsgi_app == wsgi_app_mock()

    def test_create_wsgi_app_when_enabled(self):
        with patch.object(SimpleSvn, "_is_svn_enabled") as mock_method:
            mock_method.return_value = True
            with patch("rhodecode.lib.middleware.simplesvn.SimpleSvnApp") as (wsgi_app_mock):
                config = Mock()
                wsgi_app = self.app._create_wsgi_app(repo_path="", repo_name="", config=config)

            wsgi_app_mock.assert_called_once_with(config)
            assert wsgi_app == wsgi_app_mock()


def basic_auth(username, password):
    token = b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
    return f"Basic {token}"


class TestSimpleSvnApp(object):
    data = b"<xml></xml>"
    path = SVN_REPO
    wsgi_input = io.BytesIO(data)
    environment = {
        "HTTP_DAV": (
            "http://subversion.tigris.org/xmlns/dav/svn/depth, http://subversion.tigris.org/xmlns/dav/svn/mergeinfo"
        ),
        "HTTP_USER_AGENT": "SVN/1.14.1 (x86_64-linux) serf/1.3.8",
        "REQUEST_METHOD": "OPTIONS",
        "PATH_INFO": path,
        "wsgi.input": wsgi_input,
        "CONTENT_TYPE": "text/xml",
        "CONTENT_LENGTH": "130",
        "Authorization": basic_auth(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS),
    }

    def setup_method(self, method):
        # note(marcink): this is hostname from docker compose used for testing...
        self.host = "http://svn:8090"
        base_path = get_rhodecode_repo_store_path()
        self.app = SimpleSvnApp(config={"subversion_http_server_url": self.host, "base_path": base_path})

    def test_get_request_headers_with_content_type(self):
        expected_headers = {
            "Dav": self.environment["HTTP_DAV"],
            "User-Agent": self.environment["HTTP_USER_AGENT"],
            "Content-Type": self.environment["CONTENT_TYPE"],
            "Content-Length": self.environment["CONTENT_LENGTH"],
            "Authorization": self.environment["Authorization"],
        }
        headers = self.app._get_request_headers(self.environment)
        assert headers == expected_headers

    def test_get_request_headers_without_content_type(self):
        environment = self.environment.copy()
        environment.pop("CONTENT_TYPE")
        expected_headers = {
            "Dav": environment["HTTP_DAV"],
            "Content-Length": self.environment["CONTENT_LENGTH"],
            "User-Agent": environment["HTTP_USER_AGENT"],
            "Authorization": self.environment["Authorization"],
        }
        request_headers = self.app._get_request_headers(environment)
        assert request_headers == expected_headers

    def test_get_response_headers(self):
        headers = {
            "Connection": "keep-alive",
            "Keep-Alive": "timeout=5, max=100",
            "Transfer-Encoding": "chunked",
            "Content-Encoding": "gzip",
            "MS-Author-Via": "DAV",
            "SVN-Supported-Posts": "create-txn-with-props",
        }
        expected_headers = [
            ("MS-Author-Via", "DAV"),
            ("SVN-Supported-Posts", "create-txn-with-props"),
        ]
        response_headers = self.app._get_response_headers(headers)
        assert sorted(response_headers) == sorted(expected_headers)

    @pytest.mark.parametrize(
        "svn_http_url, path_info, expected_url",
        [
            ("http://localhost:8200", "/repo_name", "http://localhost:8200/repo_name"),
            ("http://localhost:8200///", "/repo_name", "http://localhost:8200/repo_name"),
            ("http://localhost:8200", "/group/repo_name", "http://localhost:8200/group/repo_name"),
            ("http://localhost:8200/", "/group/repo_name", "http://localhost:8200/group/repo_name"),
            ("http://localhost:8200/prefix", "/repo_name", "http://localhost:8200/prefix/repo_name"),
            ("http://localhost:8200/prefix", "repo_name", "http://localhost:8200/prefix/repo_name"),
            ("http://localhost:8200/prefix", "/group/repo_name", "http://localhost:8200/prefix/group/repo_name"),
        ],
    )
    def test_get_url(self, svn_http_url, path_info, expected_url):
        url = self.app._get_url(svn_http_url, path_info)
        assert url == expected_url

    def test_call(self):
        start_response = Mock()
        response_mock = Mock()
        response_mock.headers = {
            "Content-Encoding": "gzip",
            "MS-Author-Via": "DAV",
            "SVN-Supported-Posts": "create-txn-with-props",
        }

        from rhodecode.lib.middleware.simplesvn import requests

        original_request = requests.Session.request

        with patch("rhodecode.lib.middleware.simplesvn.requests.Session.request", autospec=True) as request_mock:
            # Use side_effect to call the original method
            request_mock.side_effect = original_request
            self.app(self.environment, start_response)

        expected_url = f"{self.host.strip('/')}/{self.path}"
        expected_request_headers = {
            "Dav": self.environment["HTTP_DAV"],
            "User-Agent": self.environment["HTTP_USER_AGENT"],
            "Authorization": self.environment["Authorization"],
            "Content-Type": self.environment["CONTENT_TYPE"],
            "Content-Length": self.environment["CONTENT_LENGTH"],
        }

        # Check if the method was called
        assert request_mock.called
        assert request_mock.call_count == 1

        # Extract the session instance from the first call
        called_with_session = request_mock.call_args[0][0]

        request_mock.assert_called_once_with(
            called_with_session,
            self.environment["REQUEST_METHOD"],
            expected_url,
            data=self.data,
            headers=expected_request_headers,
            stream=False,
        )

        expected_response_headers = [
            ("SVN-Supported-Posts", "create-txn-with-props"),
            ("MS-Author-Via", "DAV"),
        ]

        # TODO: the svn doesn't have a repo for testing
        # args, _ = start_response.call_args
        # assert args[0] == '200 OK'
        # assert sorted(args[1]) == sorted(expected_response_headers)
