flyweight/test_array_processor.py at main · and47/flyweight · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
import gc
import numpy as np
import pytest
from copy import deepcopy
from stateful_refs import ArrayLightProcessor


@pytest.fixture()
def arr():
    return np.random.rand(7)


@pytest.fixture(autouse=True)
def resets_class_state():
    """Ensure isolation between tests by avoiding leakage"""
    util_reset_class_state()
    yield


def util_reset_class_state():
    ArrayLightProcessor._operation_cache.clear()
    ArrayLightProcessor._cache_stats = {"hits": 0, "misses": 0}
    ArrayLightProcessor._input_tracker._addr_to_id.clear()
    ArrayLightProcessor._input_tracker._id_to_weakref.clear()
    ArrayLightProcessor._input_tracker._next_id = 0
    gc.collect()


# test cache/memory behavior

def test_idempotent_operation_does_not_increase_cache(arr):
    fw = ArrayLightProcessor(arr)

    fw = fw.normalize()
    size_after_first = len(ArrayLightProcessor._operation_cache)

    fw = fw.normalize()  # second normalize should skip
    size_after_second = len(ArrayLightProcessor._operation_cache)

    assert size_after_first == size_after_second


def test_non_idempotent_operation_increases_cache(arr):
    fw = ArrayLightProcessor(arr)

    fw1 = fw.scale(2.0).copy()
    size_after_first = sum((len(i) for i in ArrayLightProcessor._operation_cache.values()))

    fw2 = fw.normalize().copy()
    size_after_second = sum((len(i) for i in ArrayLightProcessor._operation_cache.values()))

    assert size_after_second > size_after_first


def test_memory_cleanup_on_instance_drop(arr):
    fw = ArrayLightProcessor(arr)
    name = fw._input_id

    fw1 = fw.normalize().copy()
    fw2 = fw.scale(42).copy()
    assert len(ArrayLightProcessor._operation_cache) == 1
    assert len(ArrayLightProcessor._operation_cache[name]) == 2

    del fw, fw1, fw2
    gc.collect()

    assert len(ArrayLightProcessor._operation_cache[name]) == 0


def test_input_tracker_cleanup_on_array_drop(arr):
    the_arr = arr.copy()
    fw = ArrayLightProcessor(the_arr)
    assert len(fw._input_tracker._addr_to_id) == 1

    del the_arr, fw
    gc.collect()

    assert len(ArrayLightProcessor._input_tracker._addr_to_id) == 0


# deep copy or preserve original

def test_deepcopy_creates_new_input_and_data(arr):
    fw = ArrayLightProcessor(arr)
    fw_deep = deepcopy(fw)

    assert fw._input_id != fw_deep._input_id, "Expected different tracked input id"
    assert fw.result is not fw_deep.result, "Data should not be the same object"
    np.testing.assert_array_equal(fw.result, fw_deep.result, err_msg="Content should be the same")


def test_preserve_original_true_keeps_strong_ref(arr):
    the_arr = arr.copy()
    fw = ArrayLightProcessor(the_arr, preserve_original=True)

    ref = fw.get_original_array()
    ref_vals = ref.tolist()
    del the_arr, ref
    gc.collect()

    # still alive because strong ref
    assert fw.get_original_array() is not None
    assert ref_vals == fw.get_original_array().tolist()


# shallow copy or not preserving original:

def test_preserve_original_false_allows_gc(arr):
    the_arr = arr.copy()
    fw = ArrayLightProcessor(the_arr, preserve_original=False)

    ref = fw.get_original_array()
    assert ref is the_arr  # initially same object

    fw2 = fw.copy()
    fw2 = fw.scale(42)  # perform any operation
    del fw, the_arr, ref
    gc.collect()

    assert fw2.get_original_array() is None, "weakref should now return None"


def test_copy_shares_input_id_and_data(arr):
    fw = ArrayLightProcessor(arr)
    fw_copy = fw.copy()

    assert fw._input_id == fw_copy._input_id, "Expected the same tracked input id"
    assert fw.result is fw_copy.result, "Expected the same data reference"