Coverage for tests/test_transactions.py: 100%

1import pandas as pd

2import pytest

4from sankey_cashflow import RowLabels, Transactions

7class TestInit:

9 def test_init(self, sample_transactions):

10 assert sample_transactions.length == 2

11 assert sample_transactions.earliest_date == pd.to_datetime('2023-01-01')

12 assert sample_transactions.latest_date == pd.to_datetime('2023-01-02')

14 def test_validate_df(self, sample_transactions):

15 assert sample_transactions._validate_df() is True

17 def test_init_raises_on_non_float_amount(self, make_transactions_df, sample_row_labels, default_app_settings):

18 df = make_transactions_df([

19 {'Date': '2023-01-01', 'Category': 'Groceries', 'Amount': '40.00'},

20 ])

21 with pytest.raises(Exception):

22 Transactions(df, sample_row_labels, default_app_settings)

25class TestAudit:

27 def test_audit_no_missing_transactions(self, sample_transactions):

28 audit_data = pd.DataFrame({

29 'Date': [pd.to_datetime('2023-01-01')],

30 'Amount': [107],

31 'Description': ['Grocery Store']

32 })

33 assert sample_transactions.audit(audit_data) == ""

35 def test_audit_reports_missing_transaction(self, sample_transactions):

36 audit_data = pd.DataFrame({

37 'Date': [pd.to_datetime('2023-06-01')],

38 'Amount': [9999],

39 'Description': ['Unknown Charge']

40 })

41 report = sample_transactions.audit(audit_data)

42 assert 'Unknown Charge' in report

45class TestFilterTags:

47 def test_filter_tags_drops_matching_rows(self, sample_transactions):

48 sample_transactions.filter_tags(['Recurring'])

49 assert len(sample_transactions._df) == 1

50 assert 'Recurring' not in sample_transactions._df['Tags'].tolist()

52 def test_filter_tags_no_match_is_noop(self, sample_transactions):

53 sample_transactions.filter_tags(['NoSuchTag'])

54 assert len(sample_transactions._df) == 2

57class TestApplyLabelsBasic:

59 def test_apply_labels_resolves_default_source_target(self, sample_transactions):

60 sample_transactions.apply_labels()

61 assert sample_transactions._df.at[0, 'Source'] == 'Income'

62 assert sample_transactions._df.at[0, 'Target'] == 'Groceries'

65@pytest.fixture

66def tag_labels_data():

67 return [

68 {'Category Name': 'Food', 'Type': 'computed', 'Source': 'Income', 'Target': 'Food',

69 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

70 {'Category Name': 'Coffee Shops', 'Type': '', 'Source': 'Food', 'Target': 'Eating Out',

71 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

72 {'Category Name': 'Ford', 'Type': 'tag', 'Source': 'Automotive', 'Target': 'Ford',

73 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

74 {'Category Name': 'Automotive', 'Type': 'computed', 'Source': 'Income', 'Target': 'Automotive',

75 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

76 {'Category Name': 'Gas', 'Type': '', 'Source': 'Automotive', 'Target': 'Gas',

77 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

78 {'Category Name': 'Joe', 'Type': 's-tag', 'Source': '', 'Target': '',

79 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

80 {'Category Name': 'House', 'Type': 'computed', 'Source': 'Income', 'Target': 'House',

81 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

82 {'Category Name': '401K', 'Type': '', 'Source': 'DEDUCTIONS', 'Target': '401K',

83 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

84 ]

87@pytest.fixture

88def tag_row_labels(tag_labels_data):

89 return RowLabels(tag_labels_data)

92def _transactions_for(df_rows, row_labels, make_transactions_df, make_args, **settings_overrides):

93 from sankey_cashflow import AppSettings

94 df = make_transactions_df(df_rows)

95 settings = AppSettings(make_args(**settings_overrides))

96 return Transactions(df, row_labels, settings)

99class TestApplyLabelsTagBranches:

100

101 def test_tag_append_mode(self, tag_row_labels, make_transactions_df, make_args):

102 # 'Weekend' is not defined in the labels sheet at all (not a 'tag' or 's-tag' type row),

103 # so this exercises the plain append branch rather than s-tag/tag_override handling.

104 txn = _transactions_for(

105 [{'Date': '2023-01-01', 'Category': 'Coffee Shops', 'Tags': 'Weekend', 'Amount': 10.0}],

106 tag_row_labels, make_transactions_df, make_args, tags='Weekend'

107 )

108 txn.apply_labels()

109 assert txn._df.at[0, 'Source'] == 'Eating Out'

110 assert txn._df.at[0, 'Target'] == 'Weekend'

111

112 def test_tag_override_matched(self, tag_row_labels, make_transactions_df, make_args):

113 txn = _transactions_for(

114 [{'Date': '2023-01-01', 'Category': 'Gas', 'Tags': 'Ford', 'Amount': 10.0}],

115 tag_row_labels, make_transactions_df, make_args, tags='Ford', tag_override=True

116 )

117 txn.apply_labels()

118 assert txn._df.at[0, 'Source'] == 'Automotive'

119 assert txn._df.at[0, 'Target'] == 'Ford'

120

121 def test_tag_override_unmatched_falls_back_to_income(self, tag_row_labels, make_transactions_df, make_args):

122 txn = _transactions_for(

123 [{'Date': '2023-01-01', 'Category': 'Coffee Shops', 'Tags': 'Mystery', 'Amount': 10.0}],

124 tag_row_labels, make_transactions_df, make_args, tags='Mystery', tag_override=True

125 )

126 txn.apply_labels()

127 assert txn._df.at[0, 'Source'] == 'Income'

128 assert txn._df.at[0, 'Target'] == 'Mystery'

129

130 def test_s_tag_redirects_income_source(self, tag_row_labels, make_transactions_df, make_args):

131 txn = _transactions_for(

132 [{'Date': '2023-01-01', 'Category': 'House', 'Tags': 'Joe', 'Amount': 10.0}],

133 tag_row_labels, make_transactions_df, make_args, tags='Joe'

134 )

135 txn.apply_labels()

136 assert txn._df.at[0, 'Source'] == 'Joe'

137 assert txn._df.at[0, 'Target'] == 'House'

138

139 def test_store_match(self, tag_row_labels, make_transactions_df, make_args):

140 txn = _transactions_for(

141 [{'Date': '2023-01-01', 'Category': 'Gas', 'Description': 'Costco', 'Amount': 10.0}],

142 tag_row_labels, make_transactions_df, make_args, stores='Costco'

143 )

144 txn.apply_labels()

145 assert txn._df.at[0, 'Source'] == 'Gas'

146 assert txn._df.at[0, 'Target'] == 'Costco'

147

148 def test_recurring_redirects_income_source(self, tag_row_labels, make_transactions_df, make_args):

149 txn = _transactions_for(

150 [{'Date': '2023-01-01', 'Category': 'House', 'Amount': 10.0}],

151 tag_row_labels, make_transactions_df, make_args, recurring=True

152 )

153 txn.apply_labels()

154 assert txn._df.at[0, 'Source'] == 'Recurring'

155 assert txn._df.at[0, 'Target'] == 'House'

156 assert txn._labels_obj._digraph.has_edge('Income', 'Recurring')

157

158 def test_deductions_source_uses_description(self, tag_row_labels, make_transactions_df, make_args):

159 txn = _transactions_for(

160 [{'Date': '2023-01-01', 'Category': '401K', 'Description': 'Employer', 'Amount': 10.0}],

161 tag_row_labels, make_transactions_df, make_args

162 )

163 txn.apply_labels()

164 assert txn._df.at[0, 'Source'] == 'Employer'

165 assert txn._df.at[0, 'Target'] == '401K'

166 assert txn._df.at[0, 'Type'] == 'deduction'

167

168 def test_orphan_edge_raises(self, make_transactions_df, make_args):

169 placeholder = [{'Category Name': 'default', 'Type': 'default', 'Source': '', 'Target': '',

170 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''}]

171 isolated_labels = RowLabels(placeholder)

172 txn = _transactions_for(

173 [{'Date': '2023-01-01', 'Category': 'Mystery Category', 'Amount': 10.0}],

174 isolated_labels, make_transactions_df, make_args

175 )

176 with pytest.raises(Exception):

177 txn.apply_labels()

178

179

180class TestProcessRows:

181

182 def test_sales_tax_creates_synthetic_row_and_updates_amount(self, tag_row_labels, make_transactions_df, make_args):

183 txn = _transactions_for(

184 [{'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 40.0, 'Sales Tax': 1.5}],

185 tag_row_labels, make_transactions_df, make_args

186 )

187 txn.apply_labels()

188 txn.process_rows()

189 tax_rows = txn._df[txn._df['Target'] == 'Sales Tax']

190 assert len(tax_rows) == 1

191 assert tax_rows.iloc[0]['Amount'] == 1.5

192 assert tax_rows.iloc[0]['Source'] == 'Gas'

193 assert txn._df.at[0, 'Amount'] == 41.5

194

195 def test_separate_taxes_routes_from_income(self, tag_row_labels, make_transactions_df, make_args):

196 txn = _transactions_for(

197 [{'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 40.0, 'Sales Tax': 1.5}],

198 tag_row_labels, make_transactions_df, make_args, separate_tax=True

199 )

200 txn.apply_labels()

201 txn.process_rows()

202 tax_rows = txn._df[txn._df['Target'] == 'Sales Tax']

203 assert len(tax_rows) == 1

204 assert tax_rows.iloc[0]['Source'] == 'Income'

205 # Original amount is untouched when taxes are kept separate

206 assert txn._df.at[0, 'Amount'] == 40.0

207

208 def test_tips_creates_synthetic_row_and_updates_amount(self, tag_row_labels, make_transactions_df, make_args):

209 txn = _transactions_for(

210 [{'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 40.0, 'Tips': 5.0}],

211 tag_row_labels, make_transactions_df, make_args

212 )

213 txn.apply_labels()

214 txn.process_rows()

215 tip_rows = txn._df[txn._df['Target'] == 'Tips']

216 assert len(tip_rows) == 1

217 assert tip_rows.iloc[0]['Amount'] == 5.0

218 assert txn._df.at[0, 'Amount'] == 45.0

219

220 def test_multi_hop_dag_creates_synthetic_intermediate_row(self, make_transactions_df, make_args):

221 labels_data = [

222 {'Category Name': 'House', 'Type': 'computed', 'Source': 'Income', 'Target': 'House',

223 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

224 {'Category Name': 'Mortgage', 'Type': '', 'Source': 'House', 'Target': 'Mortgage',

225 'Classification': '', 'Link color': '', 'Node color': '', 'Comments': ''},

226 ]

227 row_labels = RowLabels(labels_data)

228 txn = _transactions_for(

229 [{'Date': '2023-01-01', 'Category': 'Mortgage', 'Amount': 1500.0}],

230 row_labels, make_transactions_df, make_args

231 )

232 txn.apply_labels()

233 txn.process_rows()

234 synthetic = txn._df[(txn._df['Source'] == 'Income') & (txn._df['Target'] == 'House')]

235 assert len(synthetic) == 1

236 assert synthetic.iloc[0]['Amount'] == 1500.0

237

238

239class TestCollapse:

240

241 def test_collapse_aggregates_shared_pairs(self, make_transactions_df, tag_row_labels, make_args):

242 txn = _transactions_for(

243 [

244 {'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 40.0, 'Source': 'Automotive', 'Target': 'Gas'},

245 {'Date': '2023-01-05', 'Category': 'Gas', 'Amount': 25.0, 'Source': 'Automotive', 'Target': 'Gas'},

246 ],

247 tag_row_labels, make_transactions_df, make_args

248 )

249 txn.collapse()

250 grouped = txn._grouped_df

251 assert len(grouped) == 1

252 assert grouped.iloc[0]['Amount'] == 65.0

253

254

255class TestSurplusDeficit:

256

257 def test_income_surplus(self, make_transactions_df, tag_row_labels, make_args):

258 txn = _transactions_for(

259 [

260 {'Date': '2023-01-01', 'Category': 'Salary', 'Amount': 1000.0, 'Source': 'Job', 'Target': 'Income'},

261 {'Date': '2023-01-02', 'Category': 'Gas', 'Amount': 400.0, 'Source': 'Income', 'Target': 'Gas'},

262 ],

263 tag_row_labels, make_transactions_df, make_args

264 )

265 # In real usage this column is created by apply_labels(), which always runs before this

266 # method in Transactions.process(). Set it directly to unit test this method in isolation.

267 txn._df['Classification'] = 'Uncategorized'

268 txn.create_surplus_deficit_flows()

269 surplus_rows = txn._df[txn._df['Target'] == 'Income Surplus']

270 assert len(surplus_rows) == 1

271 assert surplus_rows.iloc[0]['Amount'] == 600.0

272 assert surplus_rows.iloc[0]['Source'] == 'Income'

273

274 def test_income_deficit(self, make_transactions_df, tag_row_labels, make_args):

275 txn = _transactions_for(

276 [

277 {'Date': '2023-01-01', 'Category': 'Salary', 'Amount': 400.0, 'Source': 'Job', 'Target': 'Income'},

278 {'Date': '2023-01-02', 'Category': 'Gas', 'Amount': 1000.0, 'Source': 'Income', 'Target': 'Gas'},

279 ],

280 tag_row_labels, make_transactions_df, make_args

281 )

282 # In real usage this column is created by apply_labels(), which always runs before this

283 # method in Transactions.process(). Set it directly to unit test this method in isolation.

284 txn._df['Classification'] = 'Uncategorized'

285 txn.create_surplus_deficit_flows()

286 deficit_rows = txn._df[txn._df['Source'] == 'Income Deficit']

287 assert len(deficit_rows) == 1

288 assert deficit_rows.iloc[0]['Amount'] == 600.0

289 assert deficit_rows.iloc[0]['Target'] == 'Income'

290

291 def test_s_tag_surplus_kept_separate_by_default(self, make_transactions_df, tag_row_labels, make_args):

292 txn = _transactions_for(

293 [

294 {'Date': '2023-01-01', 'Category': 'Salary', 'Amount': 500.0, 'Source': 'Job', 'Target': 'Joe'},

295 {'Date': '2023-01-02', 'Category': 'Gas', 'Amount': 200.0, 'Source': 'Joe', 'Target': 'Gas'},

296 ],

297 tag_row_labels, make_transactions_df, make_args

298 )

299 # In real usage this column is created by apply_labels(), which always runs before this

300 # method in Transactions.process(). Set it directly to unit test this method in isolation.

301 txn._df['Classification'] = 'Uncategorized'

302 txn.create_surplus_deficit_flows()

303 surplus_rows = txn._df[txn._df['Target'] == 'Joe Surplus']

304 assert len(surplus_rows) == 1

305 assert surplus_rows.iloc[0]['Amount'] == 300.0

306 assert surplus_rows.iloc[0]['Source'] == 'Joe'

307

308 def test_s_tag_surplus_feeds_back_to_income(self, make_transactions_df, tag_row_labels, make_args):

309 txn = _transactions_for(

310 [

311 {'Date': '2023-01-01', 'Category': 'Salary', 'Amount': 500.0, 'Source': 'Job', 'Target': 'Joe'},

312 {'Date': '2023-01-02', 'Category': 'Gas', 'Amount': 200.0, 'Source': 'Joe', 'Target': 'Gas'},

313 ],

314 tag_row_labels, make_transactions_df, make_args, feed_in=True, tags='Joe'

315 )

316 txn._df['Classification'] = 'Uncategorized'

317 txn.create_surplus_deficit_flows()

318 surplus_rows = txn._df[(txn._df['Source'] == 'Joe') & (txn._df['Target'] == 'Income')]

319 assert len(surplus_rows) == 1

320 assert surplus_rows.iloc[0]['Amount'] == 300.0

321

322

323class TestFilterDates:

324

325 @pytest.fixture

326 def dated_transactions(self, make_transactions_df, tag_row_labels, make_args):

327 return _transactions_for(

328 [

329 {'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 10.0},

330 {'Date': '2023-02-01', 'Category': 'Gas', 'Amount': 20.0},

331 {'Date': '2023-03-01', 'Category': 'Gas', 'Amount': 30.0},

332 {'Date': '2023-04-01', 'Category': 'Gas', 'Amount': 40.0},

333 ],

334 tag_row_labels, make_transactions_df, make_args

335 )

336

337 def test_filters_to_inclusive_range(self, dated_transactions):

338 dated_transactions.filter_dates('2023-02-01', '2023-03-01')

339 assert len(dated_transactions._df) == 2

340 assert dated_transactions.earliest_date == pd.to_datetime('2023-02-01')

341 assert dated_transactions.latest_date == pd.to_datetime('2023-03-01')

342

343 def test_none_none_is_noop(self, dated_transactions):

344 dated_transactions.filter_dates(None, None)

345 assert len(dated_transactions._df) == 4

346

347 def test_open_ended_start_uses_earliest(self, dated_transactions):

348 dated_transactions.filter_dates(None, '2023-02-01')

349 assert len(dated_transactions._df) == 2

350

351 def test_start_after_end_raises(self, dated_transactions):

352 with pytest.raises(Exception):

353 dated_transactions.filter_dates('2023-04-01', '2023-01-01')

354

355 def test_empty_result_raises(self, dated_transactions):

356 with pytest.raises(Exception):

357 dated_transactions.filter_dates('2024-01-01', '2024-02-01')

358

359

360class TestDistributeAmounts:

361 """

362 In Transactions.process(), distribute_amounts() (step 2) always runs BEFORE apply_labels()

363 (step 4), so the "Classification" column apply_labels() creates does not exist yet at this

364 point in real usage. These tests deliberately do NOT pre-create that column, matching real

365 call order - this is a regression test for a bug where distribute_amounts() crashed

366 unconditionally whenever a distributed row was processed before apply_labels() had run.

367 """

368

369 def test_forward_distribution_splits_amount_and_dates(self, make_transactions_df, tag_row_labels, make_args):

370 txn = _transactions_for(

371 [{'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 120.0, 'Distribution': 3}],

372 tag_row_labels, make_transactions_df, make_args

373 )

374 txn.distribute_amounts()

375 assert len(txn._df) == 3

376 assert all(txn._df['Amount'] == 40.0)

377 dates = sorted(txn._df['Date'].tolist())

378 assert dates[0] == pd.to_datetime('2023-01-01')

379 assert dates[1] > dates[0]

380 assert dates[2] > dates[1]

381

382 def test_reverse_distribution_moves_dates_backward(self, make_transactions_df, tag_row_labels, make_args):

383 txn = _transactions_for(

384 [{'Date': '2023-03-01', 'Category': 'Gas', 'Amount': 90.0, 'Distribution': -3}],

385 tag_row_labels, make_transactions_df, make_args

386 )

387 txn.distribute_amounts()

388 assert len(txn._df) == 3

389 dates = txn._df['Date'].tolist()

390 assert min(dates) < pd.to_datetime('2023-03-01')

391

392 def test_sales_tax_distributed_alongside_amount(self, make_transactions_df, tag_row_labels, make_args):

393 txn = _transactions_for(

394 [{'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 120.0, 'Distribution': 3, 'Sales Tax': 12.0}],

395 tag_row_labels, make_transactions_df, make_args

396 )

397 txn.distribute_amounts()

398 assert all(txn._df['Sales Tax'] == 4.0)

399

400 def test_calling_twice_is_noop(self, make_transactions_df, tag_row_labels, make_args):

401 txn = _transactions_for(

402 [{'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 120.0, 'Distribution': 3}],

403 tag_row_labels, make_transactions_df, make_args

404 )

405 txn.distribute_amounts()

406 assert len(txn._df) == 3

407 txn.distribute_amounts()

408 assert len(txn._df) == 3

409

410 def test_distribution_before_apply_labels_does_not_crash(self, make_transactions_df, tag_row_labels, make_args):

411 # Regression test matching the real Transactions.process() call order: distribute_amounts()

412 # before apply_labels(), with no "Classification" column present yet on either the original

413 # or the synthetic rows it creates.

414 txn = _transactions_for(

415 [{'Date': '2023-01-01', 'Category': 'Gas', 'Amount': 90.0, 'Distribution': 3}],

416 tag_row_labels, make_transactions_df, make_args

417 )

418 assert 'Classification' not in txn._df.columns

419 txn.distribute_amounts()

420 assert 'Classification' not in txn._df.columns

421 assert len(txn._df) == 3

422 txn.apply_labels()

423 assert 'Classification' in txn._df.columns

424 assert all(txn._df['Classification'].notna())

425

426 def test_no_distribution_rows_untouched(self, sample_transactions):

427 original_len = len(sample_transactions._df)

428 sample_transactions.distribute_amounts()

429 assert len(sample_transactions._df) == original_len

430

431

432class TestFullProcessPipeline:

433

434 def test_process_marks_surplus_deficit_processed(self, sample_transactions):

435 sample_transactions.process()

436 assert sample_transactions.surplus_deficit_processed is True

437 assert sample_transactions._grouped_df is not None