#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Tests for algorithms related to association rules.
"""

import os
import pytest
import itertools
import random

from efficient_apriori.itemsets import itemsets_from_transactions


def generate_transactions(
    num_transactions, unique_items, items_row=(1, 100), seed=None
):
    """
    Generate synthetic transactions.
    """
    if seed:
        random.seed(seed)
    else:
        random.seed()

    items = list(range(unique_items))

    for transaction in range(num_transactions):
        items_this_row = random.randint(*items_row)
        yield random.sample(items, k=min(unique_items, items_this_row))


def itemsets_from_transactions_naive(transactions, min_support):
    """
    Naive algorithm used for testing only.
    """

    # Get the unique items from every transaction
    unique_items = set(k for t in transactions for k in t)
    num_transactions = len(transactions)

    # Create an output dictionary
    L = dict()

    # For every possible combination length
    for k in range(1, len(unique_items) + 1):

        # For every possible combination
        for combination in itertools.combinations(unique_items, k):

            # Naively count how many transactions contain the combination
            counts = 0
            for transaction in transactions:
                if set.issubset(set(combination), set(transaction)):
                    counts += 1

            # If the count exceeds the minimum support, add it
            if (counts / num_transactions) >= min_support:
                try:
                    L[k][tuple(sorted(list(combination)))] = counts
                except KeyError:
                    L[k] = dict()
                    L[k][tuple(sorted(list(combination)))] = counts

        try:
            L[k] = {k: v for (k, v) in sorted(L[k].items())}
            if L[k] == {}:
                del L[k]
                return L, num_transactions
        except KeyError:
            return L, num_transactions

    return L, num_transactions


input_data = [
    (
        list(
            generate_transactions(
                random.randint(5, 25),
                random.randint(1, 8),
                (1, random.randint(2, 8)),
            )
        ),
        random.randint(1, 4) / 10,
    )
    for i in range(10)
]


@pytest.mark.parametrize("transactions, min_support", input_data)
def test_itemsets_from_transactions_stochastic(transactions, min_support):
    """
    Test 50 random inputs.
    """
    result, _ = itemsets_from_transactions(list(transactions), min_support)
    naive_result, _ = itemsets_from_transactions_naive(
        list(transactions), min_support
    )

    for key in set.union(set(result.keys()), set(naive_result.keys())):
        assert result[key] == naive_result[key]


@pytest.mark.parametrize("transactions, min_support", input_data)
def test_itemsets_max_length(transactions, min_support):
    """
    The that nothing larger than max length is returned.
    """
    max_len = random.randint(1, 5)
    result, _ = itemsets_from_transactions(
        list(transactions), min_support, max_length=max_len
    )

    assert all(list(k <= max_len for k in result.keys()))


def test_itemsets_from_a_generator_callable():
    """
    Test generator feature.
    """

    def generator():
        """
        A generator for testing.
        """
        for i in range(4):
            yield [j + i for j in range(5)]

    itemsets, _ = itemsets_from_transactions(generator, min_support=3 / 4)
    assert itemsets[3] == {(2, 3, 4): 3, (3, 4, 5): 3}


def test_itemsets_from_a_file():
    """
    Test generator feature.
    """

    def file_generator(filename):
        """
        A file generator for testing.
        """

        def generate_from_file():
            with open(filename) as file:
                for line in file:
                    yield tuple(line.strip("\n").split(","))

        return generate_from_file

    base, filename = os.path.split(__file__)
    gen_obj = file_generator(os.path.join(base, "transactions.txt"))
    result, _ = itemsets_from_transactions(gen_obj, min_support=4 / 4)
    assert result[2] == {("A", "C"): 4}


if __name__ == "__main__":
    pytest.main(args=[".", "--doctest-modules", "-v"])


================================================================
====================================================================
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Tests for algorithms related to association rules.
"""

import pytest
from efficient_apriori.apriori import apriori
from efficient_apriori.rules import Rule


def test_against_R_implementation_1():
    """
    The following R-code was used:
        
    > install.packages("arules")
    > col1 = c("a", "a", "a", "b", "b", "b", "b") 
    > col2 = c("c", "c", "d", "d", "d", "c", "c") 
    > col3 = c("e", "e", "e", "e", "f", "f", "f")
    > df = data.frame(col1, col2, col3)
    > df <- data.frame(sapply(df, as.factor))
    > rules <- apriori(df, parameter = list(supp = 0.2, conf = 0.2))
    > inspect(head(rules, by = "confidence"))
    """

    transactions = [
        ("a", "c", "e"),
        ("a", "c", "e"),
        ("a", "d", "e"),
        ("b", "d", "e"),
        ("b", "d", "f"),
        ("b", "c", "f"),
        ("b", "c", "f"),
    ]

    itemsets, rules = apriori(transactions, 0.2, 0.2)

    assert Rule(("a",), ("e",)) in rules

    for rule in rules:
        if rule == Rule(("a",), ("e",)):
            assert abs(rule.support - 0.4285714) < 10e-7
            assert rule.confidence == 1

        if rule == Rule(("c", "e"), ("a",)):
            assert abs(rule.support - 0.2857143) < 10e-7
            assert rule.confidence == 1

        if rule == Rule(("e",), ("a",)):
            assert abs(rule.support - 0.4285714) < 10e-7
            assert rule.confidence == 3 / 4


def test_against_R_implementation_2():
    """
    The following R-code was used:
        
    > install.packages("arules")
    > col1 = c("b", "b", "c", "b", "a", "a", "b", "c", "b", "b", "a", "b", "a", 
    "a", "a", "c", "b", "a", "b", "b", "b", "c", "a", "c", "a", "a", "c", "a", 
    "b", "b", "a", "c") 
    > col2 = c("e", "f", "e", "e", "f", "e", "d", "f", "e", "e", "e", "d", "e", 
    "e", "f", "d", "d", "d", "e", "f", "f", "d", "d", "f", "e", "e", "f", "f", 
    "f", "d", "e", "e") 
    > col3 = c("g", "i", "j", "i", "i", "j", "i", "h", "g", "j", "g", "h", "i", 
    "h", "g", "h", "g", "j", "h", "i", "g", "g", "i", "h", "h", "h", "h", "g", 
    "j", "i", "g", "g")
    > df = data.frame(col1, col2, col3)
    > df <- data.frame(sapply(df, as.factor))
    > rules <- apriori(df, parameter = list(supp = 0.2, conf = 0.2))
    > inspect(head(rules, by = "confidence"))
    """

    transactions = [
        ("b", "e", "g"),
        ("b", "f", "i"),
        ("c", "e", "j"),
        ("b", "e", "i"),
        ("a", "f", "i"),
        ("a", "e", "j"),
        ("b", "d", "i"),
        ("c", "f", "h"),
        ("b", "e", "g"),
        ("b", "e", "j"),
        ("a", "e", "g"),
        ("b", "d", "h"),
        ("a", "e", "i"),
        ("a", "e", "h"),
        ("a", "f", "g"),
        ("c", "d", "h"),
        ("b", "d", "g"),
        ("a", "d", "j"),
        ("b", "e", "h"),
        ("b", "f", "i"),
        ("b", "f", "g"),
        ("c", "d", "g"),
        ("a", "d", "i"),
        ("c", "f", "h"),
        ("a", "e", "h"),
        ("a", "e", "h"),
        ("c", "f", "h"),
        ("a", "f", "g"),
        ("b", "f", "j"),
        ("b", "d", "i"),
        ("a", "e", "g"),
        ("c", "e", "g"),
    ]

    itemsets, rules = apriori(transactions, 0.2, 0.2)

    for rule in rules:
        if rule == Rule(("a",), ("e",)):
            assert abs(rule.support - 0.21875) < 10e-7
            assert abs(rule.confidence - 0.5833333) < 10e-7

        if rule == Rule(("e",), ("a",)):
            assert abs(rule.support - 0.21875) < 10e-7
            assert abs(rule.confidence - 0.5000000) < 10e-7


def test_against_R_implementation_3():
    """
    The following R-code was used:
        
    > install.packages("arules")
    > col1 = c("b", "b", "c", "a", "b", "b", "a", "a", "b", "b", "a", "a", "c", 
    "b", "a", "c") 
    > col2 = c("e", "d", "e", "e", "e", "e", "d", "e", "e", "e", "d", "e", "e", 
    "e", "d", "e") 
    > col3 = c("i", "g", "h", "j", "i", "g", "h", "j", "i", "g", "j", "i", "j", 
    "j", "i", "i")
    > df = data.frame(col1, col2, col3)
    > df <- data.frame(sapply(df, as.factor))
    > rules <- apriori(df, parameter = list(supp = 0.2, conf = 0.2))
    > inspect(head(rules, by = "confidence"))
    """

    transactions = [
        ("b", "e", "i"),
        ("b", "d", "g"),
        ("c", "e", "h"),
        ("a", "e", "j"),
        ("b", "e", "i"),
        ("b", "e", "g"),
        ("a", "d", "h"),
        ("a", "e", "j"),
        ("b", "e", "i"),
        ("b", "e", "g"),
        ("a", "d", "j"),
        ("a", "e", "i"),
        ("c", "e", "j"),
        ("b", "e", "j"),
        ("a", "d", "i"),
        ("c", "e", "i"),
    ]

    itemsets, rules = apriori(transactions, 0.2, 0.2)

    for rule in rules:
        if rule == Rule(("b",), ("e",)):
            assert abs(rule.support - 0.3750) < 10e-7
            assert abs(rule.confidence - 0.8571429) < 10e-7

        if rule == Rule(("i",), ("e",)):
            assert abs(rule.support - 0.3125) < 10e-7
            assert abs(rule.confidence - 0.8333333) < 10e-7

        if rule == Rule(("j",), ("e",)):
            assert abs(rule.support - 0.2500) < 10e-7
            assert abs(rule.confidence - 0.8000000) < 10e-7

        if rule == Rule(("e",), ("b",)):
            assert abs(rule.support - 0.3750) < 10e-7
            assert abs(rule.confidence - 0.5000000) < 10e-7


def test_minimal_input():
    """
    The with some minimal inputs, and make sure the correct errors are raised.
    """
    transactions = []
    itemsets, rules = apriori(transactions, 0.2, 0.2)
    assert itemsets == {} and rules == []

    with pytest.raises(ValueError):
        itemsets, rules = apriori(transactions, -0.2, 0.2)

    with pytest.raises(ValueError):
        itemsets, rules = apriori(transactions, 0.2, -0.2)

    with pytest.raises(ValueError):
        itemsets, rules = apriori(transactions, "asdf", 1)

    itemsets, rules = apriori([(1, 2), (1, 2), (1, 3)], 1, 1)
    itemsets, rules = apriori([(1, 2), (1, 2), (1, 3)], 1.0, 1.0)


if __name__ == "__main__":
    pytest.main(args=[".", "--doctest-modules", "-v"])