<commit_before>#! /usr/bin/env python3.6
# coding: utf-8
from kamera.logger import log

import os
import datetime as dt
from pathlib import Path
from types import SimpleNamespace
import shutil
from io import BytesIO
import fakeredis

import dropbox
import pytest
import pytz
from PIL import Image

from kamera import config
from kamera.task import Task

from typing import Optional, Dict


default_client_modified = dt.datetime(2000, 1, 1)


def make_image(
    changed: bool, dimensions: Optional[dropbox.files.Dimensions] = None
) -> bytes:
    dimensions = (
        (dimensions.width, dimensions.height) if dimensions is not None else (100, 100)
    )
    new_data = BytesIO()
    image = Image.new("RGB", size=dimensions, color=1 if changed else 2)
    image.save(new_data, "JPEG")
    data = new_data.getvalue()
    return data


def make_all_temp_folders(root_dir: Path) -> None:
    os.mkdir(root_dir / "Uploads")
    os.mkdir(root_dir / "Review")
    os.mkdir(root_dir / "Backup")
    os.mkdir(root_dir / "Error")


def run_task_process_entry(
    test_name: str,
    ext: str,
    root_dir: Path,
    file_name: Optional[str] = None,
    metadata: Optional[dropbox.files.PhotoMetadata] = None,
) -> None:
    in_file = root_dir / "Uploads" / f"{file_name}{ext}"
    account_id = test_name
    image = make_image(
        changed=False, dimensions=metadata.dimensions if metadata else None
    )
    with open(in_file, "wb") as file:
        file.write(image)
    dbx_entry = dropbox.files.FileMetadata(
        path_display=in_file.as_posix(), client_modified=default_client_modified
    )
    Task.dbx_cache[account_id] = MockDropbox()
    Task.settings_cache[account_id] = MockSettings(account_id)
    Task.redis_client = fakeredis.FakeStrictRedis()

    task = Task(
        account_id=account_id,
        entry=dbx_entry,
        metadata=metadata,
        review_dir=root_dir / "Review",
        backup_dir=root_dir / "Backup",
        error_dir=root_dir / "Error",
    )
    task.process_entry()


def _get_folder_contents(root_dir: Path):
    uploads_contents = list((root_dir / "Uploads").rglob("*"))
    review_contents = list((root_dir / "Review").rglob("*"))
    backup_contents = list((root_dir / "Backup").rglob("*"))
    error_contents = list((root_dir / "Error").rglob("*"))
    return uploads_contents, review_contents, backup_contents, error_contents


def assert_file_not_moved(root_dir: Path) -> None:
    uploads_contents, review_contents, backup_contents, error_contents = _get_folder_contents(
        root_dir
    )
    assert error_contents == []
    assert len(uploads_contents) == 1
    assert review_contents == []
    assert backup_contents == []


def assert_file_moved_to_review_and_backup(root_dir: Path) -> None:
    uploads_contents, review_contents, backup_contents, error_contents = _get_folder_contents(
        root_dir
    )
    assert error_contents == []
    assert uploads_contents == []
    assert len(review_contents) == 3
    assert len(backup_contents) == 3


def assert_file_moved_to_error(root_dir: Path) -> None:
    uploads_contents, review_contents, backup_contents, error_contents = _get_folder_contents(
        root_dir
    )
    assert len(error_contents) == 1
    assert uploads_contents == []
    assert review_contents == []
    assert backup_contents == []


def assert_contents_changed(root_dir: Path, subfolder: str) -> None:
    out_file = list((root_dir / subfolder).rglob("*"))[-1]
    with open(out_file, "rb") as file:
        image = file.read()
    assert image == make_image(changed=True)


def assert_contents_unchanged(root_dir: Path, subfolder: str) -> None:
    out_file = list((root_dir / subfolder).rglob("*"))[-1]
    with open(out_file, "rb") as file:
        image = file.read()
    assert image == make_image(changed=False)


@pytest.mark.parametrize("extension", config.video_extensions)
@pytest.mark.usefixtures("data_from_img_processing")
def test_video(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_video{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_unchanged(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")


@pytest.mark.parametrize("extension", config.image_extensions)
@pytest.mark.usefixtures("data_from_img_processing")
def test_image_changed(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_image_changed{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_changed(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")


@pytest.mark.parametrize("extension", config.image_extensions)
@pytest.mark.usefixtures("no_img_processing")
def test_image_unchanged(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_image_unchanged{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_unchanged(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")


@pytest.mark.parametrize("extension", config.media_extensions)
@pytest.mark.usefixtures("data_from_img_processing", "error_parse_date")
def test_error(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_error{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_error(root_dir)
    assert_contents_unchanged(root_dir, "Error")


@pytest.mark.usefixtures()
def test_unsupported_ext(tmpdir) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name="test_unsupported_ext", ext=".ext", root_dir=root_dir
    )
    assert_file_not_moved(root_dir)
    assert_contents_unchanged(root_dir, "Uploads")


@pytest.mark.usefixtures("data_from_img_processing")
def test_client_modified_date_used(tmpdir, settings) -> None:
    """datetime sent to image_processing (default_client_modified) is timezone-naive 01.01.2000 00:00
    this should be assumed to be utc. tests default timezone is "US/Eastern" (utc-05:00).
    client_modified_local should be 12.31.1999 19:00, folders created should be 1999 and 12
     """
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    client_modified_utc = default_client_modified.replace(tzinfo=dt.timezone.utc)
    client_modified_local = client_modified_utc.astimezone(
        tz=pytz.timezone("US/Eastern")
    )
    run_task_process_entry(
        test_name="test_client_modified_date_used", ext=".jpg", root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_changed(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")

    year_folder, month_folder, out_file = (root_dir / "Review").rglob("*")
    assert year_folder.name == str(client_modified_local.year)
    assert month_folder.name == settings.folder_names[client_modified_local.month]


@pytest.mark.usefixtures("data_from_img_processing")
def test_time_taken_date_used(tmpdir, settings) -> None:
    """datetime sent to image_processing (in_date_naive) is timezone-naive 01.01.2010 00:00
    this should be assumed to be utc. tests default timezone is "US/Eastern" (utc-05:00).
    client_modified_local should be 12.31.2009 19:00, folders created should be 2009 and 12
     """
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    in_date_naive = dt.datetime(2010, 1, 1, 0, 0)
    in_date_utc = in_date_naive.replace(tzinfo=dt.timezone.utc)
    in_date_local = in_date_utc.astimezone(tz=pytz.timezone("US/Eastern"))
    metadata = dropbox.files.PhotoMetadata(
        dimensions=None, location=None, time_taken=in_date_naive
    )
    run_task_process_entry(
        test_name="test_time_taken_date_used",
        ext=".jpg",
        root_dir=root_dir,
        metadata=metadata,
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_changed(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")

    year_folder, month_folder, out_file = (root_dir / "Review").rglob("*")
    assert year_folder.name == str(in_date_local.year)
    assert month_folder.name == settings.folder_names[in_date_local.month]


@pytest.mark.usefixtures("data_from_img_processing")
def test_time_taken_date_used_with_location(tmpdir, settings) -> None:
    """datetime sent to image_processing (in_date_naive) is timezone-naive 12.31.2014 23:00,
    with gps location in timezone "Europe/Paris"
    time_taken should be assumed to be utc. tests default timezone is "US/Eastern" (utc-05:00).
    in_date_local should be 01.01.2015 00:00, folders created should be 2015 and 01
     """
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    in_date_naive = dt.datetime(2014, 12, 31, 23, 0)
    in_date_utc = in_date_naive.replace(tzinfo=dt.timezone.utc)
    in_date_local = in_date_utc.astimezone(tz=pytz.timezone("Europe/Paris"))
    metadata = dropbox.files.PhotoMetadata(
        dimensions=None,
        location=dropbox.files.GpsCoordinates(latitude=48.8662694, longitude=2.3242583),
        time_taken=in_date_naive,
    )
    run_task_process_entry(
        test_name="test_time_taken_date_used_with_location",
        ext=".jpg",
        root_dir=root_dir,
        metadata=metadata,
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_changed(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")

    year_folder, month_folder, out_file = (root_dir / "Review").rglob("*")
    assert year_folder.name == str(in_date_local.year)
    assert month_folder.name == settings.folder_names[in_date_local.month]

<commit_msg>Add extension parameterization to time zone tests<commit_after>#! /usr/bin/env python3.6
# coding: utf-8
from kamera.logger import log

import os
import datetime as dt
from pathlib import Path
from types import SimpleNamespace
import shutil
from io import BytesIO
import fakeredis

import dropbox
import pytest
import pytz
from PIL import Image

from kamera import config
from kamera.task import Task

from typing import Optional, Dict


default_client_modified = dt.datetime(2000, 1, 1, 10, 30)


def make_image(
    changed: bool, dimensions: Optional[dropbox.files.Dimensions] = None
) -> bytes:
    dimensions = (
        (dimensions.width, dimensions.height) if dimensions is not None else (100, 100)
    )
    new_data = BytesIO()
    image = Image.new("RGB", size=dimensions, color=1 if changed else 2)
    image.save(new_data, "JPEG")
    data = new_data.getvalue()
    return data


def make_all_temp_folders(root_dir: Path) -> None:
    os.mkdir(root_dir / "Uploads")
    os.mkdir(root_dir / "Review")
    os.mkdir(root_dir / "Backup")
    os.mkdir(root_dir / "Error")


def run_task_process_entry(
    test_name: str,
    ext: str,
    root_dir: Path,
    file_name: Optional[str] = None,
    metadata: Optional[dropbox.files.PhotoMetadata] = None,
) -> None:
    in_file = root_dir / "Uploads" / f"{file_name}{ext}"
    account_id = test_name
    image = make_image(
        changed=False, dimensions=metadata.dimensions if metadata else None
    )
    with open(in_file, "wb") as file:
        file.write(image)
    dbx_entry = dropbox.files.FileMetadata(
        path_display=in_file.as_posix(), client_modified=default_client_modified
    )
    Task.dbx_cache[account_id] = MockDropbox()
    Task.settings_cache[account_id] = MockSettings(account_id)  # type: ignore
    Task.redis_client = fakeredis.FakeStrictRedis()

    task = Task(
        account_id=account_id,
        entry=dbx_entry,
        metadata=metadata,
        review_dir=root_dir / "Review",
        backup_dir=root_dir / "Backup",
        error_dir=root_dir / "Error",
    )
    task.process_entry()


def _get_folder_contents(root_dir: Path):
    uploads_contents = list((root_dir / "Uploads").rglob("*"))
    review_contents = list((root_dir / "Review").rglob("*"))
    backup_contents = list((root_dir / "Backup").rglob("*"))
    error_contents = list((root_dir / "Error").rglob("*"))
    return uploads_contents, review_contents, backup_contents, error_contents


def assert_file_not_moved(root_dir: Path) -> None:
    uploads_contents, review_contents, backup_contents, error_contents = _get_folder_contents(
        root_dir
    )
    assert error_contents == []
    assert len(uploads_contents) == 1
    assert review_contents == []
    assert backup_contents == []


def assert_file_moved_to_review_and_backup(root_dir: Path) -> None:
    uploads_contents, review_contents, backup_contents, error_contents = _get_folder_contents(
        root_dir
    )
    assert error_contents == []
    assert uploads_contents == []
    assert len(review_contents) == 3
    assert len(backup_contents) == 3


def assert_file_moved_to_error(root_dir: Path) -> None:
    uploads_contents, review_contents, backup_contents, error_contents = _get_folder_contents(
        root_dir
    )
    assert len(error_contents) == 1
    assert uploads_contents == []
    assert review_contents == []
    assert backup_contents == []


def assert_contents_changed(root_dir: Path, subfolder: str) -> None:
    out_file = list((root_dir / subfolder).rglob("*"))[-1]
    with open(out_file, "rb") as file:
        image = file.read()
    assert image == make_image(changed=True)


def assert_contents_unchanged(root_dir: Path, subfolder: str) -> None:
    out_file = list((root_dir / subfolder).rglob("*"))[-1]
    with open(out_file, "rb") as file:
        image = file.read()
    assert image == make_image(changed=False)


@pytest.mark.parametrize("extension", config.video_extensions)
@pytest.mark.usefixtures("data_from_img_processing")
def test_video(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_video{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_unchanged(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")


@pytest.mark.parametrize("extension", config.image_extensions)
@pytest.mark.usefixtures("data_from_img_processing")
def test_image_changed(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_image_changed{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_changed(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")


@pytest.mark.parametrize("extension", config.image_extensions)
@pytest.mark.usefixtures("no_img_processing")
def test_image_unchanged(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_image_unchanged{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)
    assert_contents_unchanged(root_dir, "Review")
    assert_contents_unchanged(root_dir, "Backup")


@pytest.mark.parametrize("extension", config.media_extensions)
@pytest.mark.usefixtures("data_from_img_processing", "error_parse_date")
def test_error(tmpdir, extension) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name=f"test_error{extension}", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_error(root_dir)
    assert_contents_unchanged(root_dir, "Error")


@pytest.mark.usefixtures()
def test_unsupported_ext(tmpdir) -> None:
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    run_task_process_entry(
        test_name="test_unsupported_ext", ext=".ext", root_dir=root_dir
    )
    assert_file_not_moved(root_dir)
    assert_contents_unchanged(root_dir, "Uploads")


@pytest.mark.parametrize("extension", config.media_extensions)
@pytest.mark.usefixtures("data_from_img_processing")
def test_client_modified_date_used(tmpdir, settings, extension) -> None:
    """datetime sent to image_processing (default_client_modified) is timezone-naive 01.01.2000 00:00
    this should be assumed to be utc. tests default timezone is "US/Eastern" (utc-05:00).
    client_modified_local should be 12.31.1999 19:00, folders created should be 1999 and 12
     """
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    client_modified_utc = default_client_modified.replace(tzinfo=dt.timezone.utc)
    client_modified_local = client_modified_utc.astimezone(
        tz=pytz.timezone("US/Eastern")
    )
    run_task_process_entry(
        test_name="test_client_modified_date_used", ext=extension, root_dir=root_dir
    )
    assert_file_moved_to_review_and_backup(root_dir)

    year_folder, month_folder, out_file = (root_dir / "Review").rglob("*")
    assert year_folder.name == str(client_modified_local.year)
    assert month_folder.name == settings.folder_names[client_modified_local.month]


@pytest.mark.parametrize("extension", config.media_extensions)
@pytest.mark.usefixtures("data_from_img_processing")
def test_time_taken_date_used(tmpdir, settings, extension) -> None:
    """datetime sent to image_processing (in_date_naive) is timezone-naive 01.01.2010 00:00
    this should be assumed to be utc. tests default timezone is "US/Eastern" (utc-05:00).
    client_modified_local should be 12.31.2009 19:00, folders created should be 2009 and 12
     """
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    in_date_naive = dt.datetime(2010, 1, 1, 0, 0)
    in_date_utc = in_date_naive.replace(tzinfo=dt.timezone.utc)
    in_date_local = in_date_utc.astimezone(tz=pytz.timezone("US/Eastern"))
    metadata = dropbox.files.PhotoMetadata(
        dimensions=None, location=None, time_taken=in_date_naive
    )
    run_task_process_entry(
        test_name="test_time_taken_date_used",
        ext=extension,
        root_dir=root_dir,
        metadata=metadata,
    )
    assert_file_moved_to_review_and_backup(root_dir)

    year_folder, month_folder, out_file = (root_dir / "Review").rglob("*")
    assert year_folder.name == str(in_date_local.year)
    assert month_folder.name == settings.folder_names[in_date_local.month]


@pytest.mark.parametrize("extension", config.media_extensions)
@pytest.mark.usefixtures("data_from_img_processing")
def test_time_taken_date_used_with_location(tmpdir, settings, extension) -> None:
    """datetime sent to image_processing (in_date_naive) is timezone-naive 12.31.2014 23:00,
    with gps location in timezone "Europe/Paris"
    time_taken should be assumed to be utc. tests default timezone is "US/Eastern" (utc-05:00).
    in_date_local should be 01.01.2015 00:00, folders created should be 2015 and 01
     """
    root_dir = Path(tmpdir)
    make_all_temp_folders(root_dir)
    in_date_naive = dt.datetime(2014, 12, 31, 23, 0)
    in_date_utc = in_date_naive.replace(tzinfo=dt.timezone.utc)
    in_date_local = in_date_utc.astimezone(tz=pytz.timezone("Europe/Paris"))
    metadata = dropbox.files.PhotoMetadata(
        dimensions=None,
        location=dropbox.files.GpsCoordinates(latitude=48.8662694, longitude=2.3242583),
        time_taken=in_date_naive,
    )
    run_task_process_entry(
        test_name="test_time_taken_date_used_with_location",
        ext=extension,
        root_dir=root_dir,
        metadata=metadata,
    )
    assert_file_moved_to_review_and_backup(root_dir)

    year_folder, month_folder, out_file = (root_dir / "Review").rglob("*")
    assert year_folder.name == str(in_date_local.year)
    assert month_folder.name == settings.folder_names[in_date_local.month]


{'fraction_non_alphanumeric': 0.10531866807292452, 'fraction_numerical': 0.016372859223544773, 'mean_word_length': 4.467325645249863, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3697984', 'n_tokens_mistral': 7037, 'n_tokens_neox': 6772, 'n_words': 1201}