path: root/test/test_static.py

import os
import socket

import pytest
from unit.applications.proto import TestApplicationProto
from unit.option import option
from unit.utils import waitforfiles


class TestStatic(TestApplicationProto):
    prerequisites = {}

    def setup_method(self):
        os.makedirs(f'{option.temp_dir}/assets/dir')
        with open(f'{option.temp_dir}/assets/index.html', 'w') as index, open(
            f'{option.temp_dir}/assets/README', 'w'
        ) as readme, open(
            f'{option.temp_dir}/assets/log.log', 'w'
        ) as log, open(
            f'{option.temp_dir}/assets/dir/file', 'w'
        ) as file:
            index.write('0123456789')
            readme.write('readme')
            log.write('[debug]')
            file.write('blah')

        self._load_conf(
            {
                "listeners": {"*:7080": {"pass": "routes"}},
                "routes": [
                    {"action": {"share": f'{option.temp_dir}/assets$uri'}}
                ],
                "settings": {
                    "http": {
                        "static": {
                            "mime_types": {"text/plain": [".log", "README"]}
                        }
                    }
                },
            }
        )

    def test_static_index(self, temp_dir):
        def set_index(index):
            assert 'success' in self.conf(
                {"share": f'{temp_dir}/assets$uri', "index": index},
                'routes/0/action',
            ), 'configure index'

        set_index('README')
        assert self.get()['body'] == 'readme', 'index'

        self.conf_delete('routes/0/action/index')
        assert self.get()['body'] == '0123456789', 'delete index'

        set_index('')
        assert self.get()['status'] == 404, 'index empty'

    def test_static_index_default(self):
        assert self.get(url='/index.html')['body'] == '0123456789', 'index'
        assert self.get(url='/')['body'] == '0123456789', 'index 2'
        assert self.get(url='//')['body'] == '0123456789', 'index 3'
        assert self.get(url='/.')['body'] == '0123456789', 'index 4'
        assert self.get(url='/./')['body'] == '0123456789', 'index 5'
        assert self.get(url='/?blah')['body'] == '0123456789', 'index vars'
        assert self.get(url='/#blah')['body'] == '0123456789', 'index anchor'
        assert self.get(url='/dir/')['status'] == 404, 'index not found'

        resp = self.get(url='/index.html/')
        assert resp['status'] == 404, 'index not found 2 status'
        assert (
            resp['headers']['Content-Type'] == 'text/html'
        ), 'index not found 2 Content-Type'

    def test_static_index_invalid(self, skip_alert, temp_dir):
        skip_alert(r'failed to apply new conf')

        def check_index(index):
            assert 'error' in self.conf(
                {"share": f'{temp_dir}/assets$uri', "index": index},
                'routes/0/action',
            )

        check_index({})
        check_index(['index.html', '$blah'])

    def test_static_large_file(self, temp_dir):
        file_size = 32 * 1024 * 1024
        with open(f'{temp_dir}/assets/large', 'wb') as f:
            f.seek(file_size - 1)
            f.write(b'\0')

        assert (
            len(self.get(url='/large', read_buffer_size=1024 * 1024)['body'])
            == file_size
        ), 'large file'

    def test_static_etag(self, temp_dir):
        etag = self.get(url='/')['headers']['ETag']
        etag_2 = self.get(url='/README')['headers']['ETag']

        assert etag != etag_2, 'different ETag'
        assert etag == self.get(url='/')['headers']['ETag'], 'same ETag'

        with open(f'{temp_dir}/assets/index.html', 'w') as f:
            f.write('blah')

        assert etag != self.get(url='/')['headers']['ETag'], 'new ETag'

    def test_static_redirect(self):
        resp = self.get(url='/dir')
        assert resp['status'] == 301, 'redirect status'
        assert resp['headers']['Location'] == '/dir/', 'redirect Location'
        assert 'Content-Type' not in resp['headers'], 'redirect Content-Type'

    def test_static_space_in_name(self, temp_dir):
        assets_dir = f'{temp_dir}/assets'

        os.rename(
            f'{assets_dir}/dir/file',
            f'{assets_dir}/dir/fi le',
        )
        assert waitforfiles(f'{assets_dir}/dir/fi le')
        assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'

        os.rename(f'{assets_dir}/dir', f'{assets_dir}/di r')
        assert waitforfiles(f'{assets_dir}/di r/fi le')
        assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'

        os.rename(f'{assets_dir}/di r', f'{assets_dir}/ di r ')
        assert waitforfiles(f'{assets_dir}/ di r /fi le')
        assert (
            self.get(url='/ di r /fi le')['body'] == 'blah'
        ), 'dir name enclosing'

        assert (
            self.get(url='/%20di%20r%20/fi le')['body'] == 'blah'
        ), 'dir encoded'
        assert (
            self.get(url='/ di r %2Ffi le')['body'] == 'blah'
        ), 'slash encoded'
        assert self.get(url='/ di r /fi%20le')['body'] == 'blah', 'file encoded'
        assert (
            self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah'
        ), 'encoded'
        assert (
            self.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body']
            == 'blah'
        ), 'encoded 2'

        os.rename(
            f'{assets_dir}/ di r /fi le',
            f'{assets_dir}/ di r / fi le ',
        )
        assert waitforfiles(f'{assets_dir}/ di r / fi le ')
        assert (
            self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah'
        ), 'file name enclosing'

        try:
            open(f'{temp_dir}/ф а', 'a').close()
            utf8 = True

        except KeyboardInterrupt:
            raise

        except:
            utf8 = False

        if utf8:
            os.rename(
                f'{assets_dir}/ di r / fi le ',
                f'{assets_dir}/ di r /фа йл',
            )
            assert waitforfiles(f'{assets_dir}/ di r /фа йл')
            assert (
                self.get(url='/ di r /фа йл')['body'] == 'blah'
            ), 'file name 2'

            os.rename(
                f'{assets_dir}/ di r ',
                f'{assets_dir}/ди ректория',
            )
            assert waitforfiles(f'{assets_dir}/ди ректория/фа йл')
            assert (
                self.get(url='/ди ректория/фа йл')['body'] == 'blah'
            ), 'dir name 2'

    def test_static_unix_socket(self, temp_dir):
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.bind(f'{temp_dir}/assets/unix_socket')

        assert self.get(url='/unix_socket')['status'] == 404, 'socket'

        sock.close()

    def test_static_unix_fifo(self, temp_dir):
        os.mkfifo(f'{temp_dir}/assets/fifo')

        assert self.get(url='/fifo')['status'] == 404, 'fifo'

    def test_static_method(self):
        resp = self.head()
        assert resp['status'] == 200, 'HEAD status'
        assert resp['body'] == '', 'HEAD empty body'

        assert self.delete()['status'] == 405, 'DELETE'
        assert self.post()['status'] == 405, 'POST'
        assert self.put()['status'] == 405, 'PUT'

    def test_static_path(self):
        assert self.get(url='/dir/../dir/file')['status'] == 200, 'relative'

        assert self.get(url='./')['status'] == 400, 'path invalid'
        assert self.get(url='../')['status'] == 400, 'path invalid 2'
        assert self.get(url='/..')['status'] == 400, 'path invalid 3'
        assert self.get(url='../assets/')['status'] == 400, 'path invalid 4'
        assert self.get(url='/../assets/')['status'] == 400, 'path invalid 5'

    def test_static_two_clients(self):
        sock = self.get(no_recv=True)
        sock2 = self.get(no_recv=True)

        assert sock.recv(1) == b'H', 'client 1'
        assert sock2.recv(1) == b'H', 'client 2'
        assert sock.recv(1) == b'T', 'client 1 again'
        assert sock2.recv(1) == b'T', 'client 2 again'

        sock.close()
        sock2.close()

    def test_static_mime_types(self):
        assert 'success' in self.conf(
            {
                "text/x-code/x-blah/x-blah": "readme",
                "text/plain": [".html", ".log", "file"],
            },
            'settings/http/static/mime_types',
        ), 'configure mime_types'

        assert (
            self.get(url='/README')['headers']['Content-Type']
            == 'text/x-code/x-blah/x-blah'
        ), 'mime_types string case insensitive'
        assert (
            self.get(url='/index.html')['headers']['Content-Type']
            == 'text/plain'
        ), 'mime_types html'
        assert (
            self.get(url='/')['headers']['Content-Type'] == 'text/plain'
        ), 'mime_types index default'
        assert (
            self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
        ), 'mime_types file in dir'

    def test_static_mime_types_partial_match(self):
        assert 'success' in self.conf(
            {
                "text/x-blah": ["ile", "fil", "f", "e", ".file"],
            },
            'settings/http/static/mime_types',
        ), 'configure mime_types'
        assert 'Content-Type' not in self.get(url='/dir/file'), 'partial match'

    def test_static_mime_types_reconfigure(self):
        assert 'success' in self.conf(
            {
                "text/x-code": "readme",
                "text/plain": [".html", ".log", "file"],
            },
            'settings/http/static/mime_types',
        ), 'configure mime_types'

        assert self.conf_get('settings/http/static/mime_types') == {
            'text/x-code': 'readme',
            'text/plain': ['.html', '.log', 'file'],
        }, 'mime_types get'
        assert (
            self.conf_get('settings/http/static/mime_types/text%2Fx-code')
            == 'readme'
        ), 'mime_types get string'
        assert self.conf_get(
            'settings/http/static/mime_types/text%2Fplain'
        ) == ['.html', '.log', 'file'], 'mime_types get array'
        assert (
            self.conf_get('settings/http/static/mime_types/text%2Fplain/1')
            == '.log'
        ), 'mime_types get array element'

        assert 'success' in self.conf_delete(
            'settings/http/static/mime_types/text%2Fplain/2'
        ), 'mime_types remove array element'
        assert (
            'Content-Type' not in self.get(url='/dir/file')['headers']
        ), 'mime_types removed'

        assert 'success' in self.conf_post(
            '"file"', 'settings/http/static/mime_types/text%2Fplain'
        ), 'mime_types add array element'
        assert (
            self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
        ), 'mime_types reverted'

        assert 'success' in self.conf(
            '"file"', 'settings/http/static/mime_types/text%2Fplain'
        ), 'configure mime_types update'
        assert (
            self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
        ), 'mime_types updated'
        assert (
            'Content-Type' not in self.get(url='/log.log')['headers']
        ), 'mime_types updated 2'

        assert 'success' in self.conf(
            '".log"', 'settings/http/static/mime_types/text%2Fblahblahblah'
        ), 'configure mime_types create'
        assert (
            self.get(url='/log.log')['headers']['Content-Type']
            == 'text/blahblahblah'
        ), 'mime_types create'

    def test_static_mime_types_correct(self):
        assert 'error' in self.conf(
            {"text/x-code": "readme", "text/plain": "readme"},
            'settings/http/static/mime_types',
        ), 'mime_types same extensions'
        assert 'error' in self.conf(
            {"text/x-code": [".h", ".c"], "text/plain": ".c"},
            'settings/http/static/mime_types',
        ), 'mime_types same extensions array'
        assert 'error' in self.conf(
            {
                "text/x-code": [".h", ".c", "readme"],
                "text/plain": "README",
            },
            'settings/http/static/mime_types',
        ), 'mime_types same extensions case insensitive'

    @pytest.mark.skip('not yet')
    def test_static_mime_types_invalid(self, temp_dir):
        assert 'error' in self.http(
            b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r
Host: localhost\r
Connection: close\r
Content-Length: 6\r
\r
\"blah\"""",
            raw_resp=True,
            raw=True,
            sock_type='unix',
            addr=f'{temp_dir}/control.unit.sock',
        ), 'mime_types invalid'