# Copyright (c) 2024, Thomas Goirand <zigo@debian.org>
#
# 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
#
#     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 collections
import filecmp
import mock
import os
import tempfile
import unittest
import shutil
import sys

from test.debug_logger import debug_logger
from swift.cli import drive_full_checker

GiB = 1024 * 1024 * 1024

if sys.version_info[0] == 2:
    # os.statvfs is removed from Python >= 3
    freespace_func = 'os.statvfs'
else:
    # shutil.disk_usage doesn't exist in Python <= 2
    freespace_func = 'shutil.disk_usage'


class TestContainerDeleter(unittest.TestCase):
    def setUp(self):
        self.logger = debug_logger()

    def _write_rsyncd_conf(self, path, max_conn):
        rsyncdconf = """pid file = /var/run/rsyncd.pid
uid = nobody
gid = nobody
use chroot = no
log format = %t %a %m %f %b
syslog facility = local3
timeout = 300
address = 192.168.100.2

[ account_sdb ]
path = /srv/node
read only = false
write only = no
list = yes
uid = swift
gid = swift
incoming chmod = Du=rwx,g=rx,o=rx,Fu=rw,g=r,o=r
outgoing chmod = Du=rwx,g=rx,o=rx,Fu=rw,g=r,o=r
max connections = {max_conn}
timeout = 0
lock file = /var/lock/account_sdb.lock

[ container_sdb ]
path = /srv/node
read only = false
write only = no
list = yes
uid = swift
gid = swift
incoming chmod = Du=rwx,g=rx,o=rx,Fu=rw,g=r,o=r
outgoing chmod = Du=rwx,g=rx,o=rx,Fu=rw,g=r,o=r
max connections = {max_conn}
timeout = 0
lock file = /var/lock/container_sdb.lock

[ object_sdb ]
path = /srv/node
read only = false
write only = no
list = yes
uid = swift
gid = swift
incoming chmod = Du=rwx,g=rx,o=rx,Fu=rw,g=r,o=r
outgoing chmod = Du=rwx,g=rx,o=rx,Fu=rw,g=r,o=r
max connections = {max_conn}
timeout = 0
lock file = /var/lock/object_sdb.lock

"""
        f = os.open(path, os.O_RDWR | os.O_CREAT)
        if sys.version_info[0] == 2:
            os.write(f, rsyncdconf.format(max_conn=max_conn))
        else:
            os.write(f, bytes(rsyncdconf.format(max_conn=max_conn), 'utf-8'))
        os.close(f)

    @mock.patch.object(drive_full_checker, 'ismount', return_value=True)
    @mock.patch(freespace_func)
    def test_drive_full(self, mock_freespace_func, os_path_ismount):
        # Create a temp folder to run our tests
        tmpdirname = tempfile.mkdtemp()

        storagepath = tmpdirname + '/srvnode'
        os.mkdir(storagepath)
        os.mkdir(storagepath + '/sdb')

        rsyncdpath = tmpdirname + "/rsyncd.conf"

        # Write a first rsyncd.conf with 8 connections for a,c,o
        self._write_rsyncd_conf(rsyncdpath, 8)

        if sys.version_info[0] == 2:
            retval = collections.namedtuple('statvfs_result',
                                            'f_bsize f_bavail')
            mock_freespace_func.return_value = retval(10, 10)
        else:
            retval = collections.namedtuple('usage', 'total used free')
            # This says: 10 bytes remaining
            mock_freespace_func.return_value = retval(10, 10, 10)

        drive_full_checker.configure_rsyncd_conf(10 * GiB, 8, ' account_{} ',
                                                 10 * GiB, 8, ' container_{} ',
                                                 10 * GiB, 8, ' object_{} ',
                                                 storagepath, rsyncdpath,
                                                 self.logger)

        should_be_rsyncdconf = tmpdirname + "/rsyncd_should_be.conf"
        self._write_rsyncd_conf(should_be_rsyncdconf, -1)

        # Assert that rsyncd.conf and rsyncd_should_be.conf are the same
        self.assertTrue(filecmp.cmp(rsyncdpath, should_be_rsyncdconf))

        shutil.rmtree(tmpdirname)

    @mock.patch.object(drive_full_checker, 'ismount', return_value=True)
    @mock.patch(freespace_func)
    def test_drive_with_space(self, mock_freespace_func, os_path_ismount):
        # Create a temp folder to run our tests
        tmpdirname = tempfile.mkdtemp()

        storagepath = tmpdirname + '/srvnode'
        os.mkdir(storagepath)
        os.mkdir(storagepath + '/sdb')

        rsyncdpath = tmpdirname + "/rsyncd.conf"

        # Write a first rsyncd.conf with 8 connections for a,c,o
        self._write_rsyncd_conf(rsyncdpath, -1)

        if sys.version_info[0] == 2:
            retval = collections.namedtuple('statvfs_result',
                                            'f_bsize f_bavail')
            mock_freespace_func.return_value = retval(10 * GiB, 10 * GiB)
        else:
            retval = collections.namedtuple('usage', 'total used free')
            # This says: 10 bytes remaining
            mock_freespace_func.return_value = retval(10 * GiB,
                                                      10 * GiB,
                                                      10 * GiB)

        drive_full_checker.configure_rsyncd_conf(10, 8, ' account_{} ',
                                                 10, 8, ' container_{} ',
                                                 10, 8, ' object_{} ',
                                                 storagepath, rsyncdpath,
                                                 self.logger)

        should_be_rsyncdconf = tmpdirname + "/rsyncd_should_be.conf"
        self._write_rsyncd_conf(should_be_rsyncdconf, 8)

        # Assert that rsyncd.conf and rsyncd_should_be.conf are the same
        self.assertTrue(filecmp.cmp(rsyncdpath, should_be_rsyncdconf))

        shutil.rmtree(tmpdirname)