# 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 os

import pytest

from rhodecode.lib.ext_json import json
from rhodecode.model.auth_token import AuthTokenModel
from rhodecode.model.db import Session, FileStore, Repository, User
from rhodecode.apps.file_store import utils as store_utils
from rhodecode.apps.file_store import config_keys

from rhodecode.tests import TestController
from rhodecode.tests.routes import route_path


class TestFileStoreViews(TestController):
    @pytest.fixture()
    def create_artifact_factory(self, tmpdir, ini_settings):
        def factory(user_id, content, f_name="example.txt"):
            config = ini_settings
            config[config_keys.backend_type] = config_keys.backend_legacy_filesystem

            f_store = store_utils.get_filestore_backend(config)

            filesystem_file = os.path.join(str(tmpdir), f_name)
            with open(filesystem_file, "wt") as f:
                f.write(content)

            with open(filesystem_file, "rb") as f:
                store_uid, metadata = f_store.store(f_name, f, metadata={"filename": f_name})
            os.remove(filesystem_file)

            entry = FileStore.create(
                file_uid=store_uid,
                filename=metadata["filename"],
                file_hash=metadata["sha256"],
                file_size=metadata["size"],
                file_display_name="file_display_name",
                file_description="repo artifact `{}`".format(metadata["filename"]),
                check_acl=True,
                user_id=user_id,
            )
            Session().add(entry)
            Session().commit()
            return entry

        return factory

    @pytest.mark.parametrize(
        "fid, content, exists",
        [
            ("abcde-0.jpg", "xxxxx", True),
            ("abcde-0.exe", "1234567", True),
            ("abcde-0.jpg", "xxxxx", False),
        ],
    )
    def test_get_files_from_store(self, fid, content, exists, tmpdir, user_util, ini_settings):
        user = self.log_user()
        user_id = user["user_id"]
        repo_id = user_util.create_repo().repo_id

        config = ini_settings
        config[config_keys.backend_type] = config_keys.backend_legacy_filesystem

        store_uid = fid

        if exists:
            status = 200
            f_store = store_utils.get_filestore_backend(config)
            filesystem_file = os.path.join(str(tmpdir), fid)
            with open(filesystem_file, "wt") as f:
                f.write(content)

            with open(filesystem_file, "rb") as f:
                store_uid, metadata = f_store.store(fid, f, metadata={"filename": fid})
            os.remove(filesystem_file)

            entry = FileStore.create(
                file_uid=store_uid,
                filename=metadata["filename"],
                file_hash=metadata["sha256"],
                file_size=metadata["size"],
                file_display_name="file_display_name",
                file_description="repo artifact `{}`".format(metadata["filename"]),
                check_acl=True,
                user_id=user_id,
                scope_repo_id=repo_id,
            )
            Session().add(entry)
            Session().commit()

        else:
            status = 404

        response = self.app.get(route_path("download_file", fid=store_uid), status=status)

        if exists:
            assert response.text == content

            metadata = f_store.get_metadata(store_uid)

            assert "size" in metadata

    def test_upload_files_without_content_to_store(self):
        self.log_user()
        response = self.app.post(route_path("upload_file"), params={"csrf_token": self.csrf_token}, status=200)

        assert response.json == {"error": "store_file data field is missing", "access_path": None, "store_fid": None}

    def test_upload_files_bogus_content_to_store(self):
        self.log_user()
        response = self.app.post(
            route_path("upload_file"), params={"csrf_token": self.csrf_token, "store_file": "bogus"}, status=200
        )

        assert response.json == {
            "error": "filename cannot be read from the data field",
            "access_path": None,
            "store_fid": None,
        }

    def test_upload_content_to_store(self):
        self.log_user()
        response = self.app.post(
            route_path("upload_file"),
            upload_files=[("store_file", b"myfile.txt", b"SOME CONTENT")],
            params={"csrf_token": self.csrf_token},
            status=200,
        )

        assert response.json["store_fid"]

    def test_download_file_non_scoped(self, user_util, create_artifact_factory):
        user = self.log_user()
        user_id = user["user_id"]
        content = "HELLO MY NAME IS ARTIFACT !"

        artifact = create_artifact_factory(user_id, content)
        file_uid = artifact.file_uid
        response = self.app.get(route_path("download_file", fid=file_uid), status=200)
        assert response.text == content

        # log-in to new user and test download again
        user = user_util.create_user(password="qweqwe")
        self.log_user(user.username, "qweqwe")
        response = self.app.get(route_path("download_file", fid=file_uid), status=200)
        assert response.text == content

    def test_download_file_scoped_to_repo(self, user_util, create_artifact_factory):
        user = self.log_user()
        user_id = user["user_id"]
        content = "HELLO MY NAME IS ARTIFACT !"

        artifact = create_artifact_factory(user_id, content)
        # bind to repo
        repo = user_util.create_repo()
        repo_id = repo.repo_id
        artifact.scope_repo_id = repo_id
        Session().add(artifact)
        Session().commit()

        file_uid = artifact.file_uid
        response = self.app.get(route_path("download_file", fid=file_uid), status=200)
        assert response.text == content

        # log-in to new user and test download again
        user = user_util.create_user(password="qweqwe")
        self.log_user(user.username, "qweqwe")
        response = self.app.get(route_path("download_file", fid=file_uid), status=200)
        assert response.text == content

        # forbid user the rights to repo
        repo = Repository.get(repo_id)
        user_util.grant_user_permission_to_repo(repo, user, "repository.none")
        self.app.get(route_path("download_file", fid=file_uid), status=404)

    def test_download_file_scoped_to_user(self, user_util, create_artifact_factory):
        user = self.log_user()
        user_id = user["user_id"]
        content = "HELLO MY NAME IS ARTIFACT !"

        artifact = create_artifact_factory(user_id, content)
        # bind to user
        user = user_util.create_user(password="qweqwe")

        artifact.scope_user_id = user.user_id
        Session().add(artifact)
        Session().commit()

        # artifact creator doesn't have access since it's bind to another user
        file_uid = artifact.file_uid
        self.app.get(route_path("download_file", fid=file_uid), status=404)

        # log-in to new user and test download again, should be ok since we're bind to this artifact
        self.log_user(user.username, "qweqwe")
        response = self.app.get(route_path("download_file", fid=file_uid), status=200)
        assert response.text == content

    def test_download_file_scoped_to_repo_with_bad_token(self, user_util, create_artifact_factory):
        user_id = User.get_first_super_admin().user_id
        content = "HELLO MY NAME IS ARTIFACT !"

        artifact = create_artifact_factory(user_id, content)
        # bind to repo
        repo = user_util.create_repo()
        repo_id = repo.repo_id
        artifact.scope_repo_id = repo_id
        Session().add(artifact)
        Session().commit()

        file_uid = artifact.file_uid
        self.app.get(route_path("download_file_by_token", _auth_token="bogus", fid=file_uid), status=302)

    def test_download_file_scoped_to_repo_with_token(self, user_util, create_artifact_factory):
        user = User.get_first_super_admin()
        AuthTokenModel().create(user, "test artifact token", role=AuthTokenModel.cls.ROLE_ARTIFACT_DOWNLOAD)

        user = User.get_first_super_admin()
        artifact_token = user.artifact_token

        user_id = User.get_first_super_admin().user_id
        content = "HELLO MY NAME IS ARTIFACT !"

        artifact = create_artifact_factory(user_id, content)
        # bind to repo
        repo = user_util.create_repo()
        repo_id = repo.repo_id
        artifact.scope_repo_id = repo_id
        Session().add(artifact)
        Session().commit()

        file_uid = artifact.file_uid
        response = self.app.get(
            route_path("download_file_by_token", _auth_token=artifact_token, fid=file_uid), status=200
        )
        assert response.text == content
