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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179 | import pytest
from aave_compress.engine import LEVELS, CompressionResult, compress
def test_levels_registry():
assert set(LEVELS) == {"light", "medium", "full"}
assert LEVELS["light"] == ["zero_copula"]
def test_light_applies_only_zero_copula():
result = compress("The server is down", "light")
assert result.compressed == "The server down"
assert result.markers_applied == ["zero_copula"]
assert result.original == "The server is down"
assert result.level == "light"
def test_light_skips_finna():
result = compress("I am about to deploy", "light")
assert result.compressed == "I am about to deploy"
assert result.markers_applied == []
def test_medium_applies_expected_subset():
text = (
"I am about to deploy because the API is constantly timing out "
"and the server is down"
)
result = compress(text, "medium")
assert result.compressed == (
"I'm finna deploy because the API steady timin out and the server down"
)
assert result.markers_applied == [
"prospective_finna",
"continuative_steady",
"zero_copula",
]
def test_medium_skips_habitual_be():
result = compress("The build usually fails", "medium")
assert result.compressed == "The build usually fails"
assert result.markers_applied == []
def test_full_habitual_and_zero_copula_composition():
result = compress("This endpoint usually times out when traffic is high", "full")
assert result.compressed == "This endpoint be timin out when traffic high"
assert result.markers_applied == ["habitual_be", "zero_copula"]
def test_full_completive_and_finna_composition():
result = compress(
"I have already finished testing and I am about to deploy", "full"
)
assert result.compressed == "I done tested and I'm finna deploy"
assert result.markers_applied == ["completive_done", "prospective_finna"]
def test_full_idempotent():
text = (
"This endpoint usually times out when traffic is high and "
"I am about to restart it because it is constantly crashing"
)
first = compress(text, "full")
second = compress(first.compressed, "full")
assert second.compressed == first.compressed
assert second.markers_applied == []
def test_result_type():
assert isinstance(compress("He is tired", "light"), CompressionResult)
def test_unknown_level_raises():
with pytest.raises(ValueError):
compress("He is tired", "bogus_level")
# --- strip_filler / tokenizer gating upgrades ---
def _word_count(s: str) -> int:
# deterministic stand-in tokenizer for testing the gate LOGIC only
return len(s.split())
def test_strip_filler_runs_filler_pass_then_markers():
result = compress(
"I really just want to know if the server is down",
"light",
strip_filler=True,
)
assert result.compressed == "the server down"
assert result.filler_stripped is True
assert result.markers_applied == ["zero_copula"]
def test_strip_filler_false_by_default_and_new_fields_default():
result = compress("The server is down", "light")
assert result.compressed == "The server down"
assert result.filler_stripped is False
assert result.gated is False
def test_gate_without_tokenizer_raises():
with pytest.raises(ValueError, match="tokenizer"):
compress("The server is down", "full", gate=True)
def test_gate_reverts_non_improving_marker():
# habitual_be maps "usually times out" (3 words) -> "be timin out"
# (3 words): no strict win under a word-count tokenizer, so the gate
# must revert it; zero_copula strictly shrinks, so it survives.
text = "This endpoint usually times out when traffic is high"
ungated = compress(text, "full")
assert "habitual_be" in ungated.markers_applied
gated = compress(text, "full", tokenizer=_word_count, gate=True)
assert gated.gated is True
assert "habitual_be" not in gated.markers_applied
assert gated.markers_applied == ["zero_copula"]
assert gated.compressed == "This endpoint usually times out when traffic high"
def test_gate_reverts_step_that_increases_count():
# tokenizer under which the finna form is MORE expensive than the
# original — the gated run must not apply it, the ungated run must.
def finna_hostile(s: str) -> int:
return len(s.split()) + (10 if "finna" in s else 0)
text = "I am about to deploy"
ungated = compress(text, "medium")
assert ungated.compressed == "I'm finna deploy"
gated = compress(text, "medium", tokenizer=finna_hostile, gate=True)
assert gated.compressed == "I am about to deploy"
assert gated.markers_applied == []
def test_gate_applies_to_filler_strip_too():
# tokenizer under which removing "really" is penalized: the filler
# step must be reverted and filler_stripped must stay False.
def really_friendly(s: str) -> int:
return 1 if "really" in s else 100
result = compress(
"This really works",
"light",
strip_filler=True,
tokenizer=really_friendly,
gate=True,
)
assert result.compressed == "This really works"
assert result.filler_stripped is False
def test_idempotent_with_new_options():
text = (
"Hey, so I really just want to know if the login endpoint usually "
"times out when the database is slow"
)
first = compress(
text, "full", strip_filler=True, tokenizer=_word_count, gate=True
)
second = compress(
first.compressed, "full", strip_filler=True, tokenizer=_word_count, gate=True
)
assert second.compressed == first.compressed
def test_backward_compat_exact_output():
# existing call shape, no new kwargs: byte-identical to the old engine
result = compress(
"This endpoint usually times out when traffic is high.", "full"
)
assert result.compressed == "This endpoint be timin out when traffic high."
assert result.markers_applied == ["habitual_be", "zero_copula"]
assert result.filler_stripped is False
assert result.gated is False
|