personal/monzo-ingestion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Correct tests

Browse Source
This commit is contained in:
Joshua Coles 2025-12-28 11:57:12 +00:00
parent 0d564ff299
commit a2ba83e6f8
1 changed files with 59 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
src/ingestion/db.rs
View File

@ -74,16 +74,16 @@ async fn update_expenditures(
.flat_map(|i| &i.contained_expenditures)
.cloned(),
)
.on_conflict(
OnConflict::columns(vec![
expenditure::Column::TransactionId,
expenditure::Column::Category,
])
.update_columns(expenditure::Column::iter())
.to_owned(),
)
.exec(tx)
.await?;
.on_conflict(
OnConflict::columns(vec![
expenditure::Column::TransactionId,
expenditure::Column::Category,
])
.update_columns(expenditure::Column::iter())
.to_owned(),
)
.exec(tx)
.await?;
Ok(())
}
@ -119,7 +119,7 @@ async fn whittle_insertions<'a>(
.select_only()
.columns([transaction::Column::IdentityHash])
.filter(transaction::Column::IdentityHash.is_not_null())
.into_tuple::<(i64, )>()
.into_tuple::<(i64,)>()
.all(tx)
.await?;
@ -133,7 +133,7 @@ async fn whittle_insertions<'a>(
let hash = i.identity_hash;
!existing_hashes
.iter()
.any(|(existing_hash, )| *existing_hash == hash)
.any(|(existing_hash,)| *existing_hash == hash)
})
.collect::<Vec<_>>();
@ -159,16 +159,17 @@ async fn notify_new_transactions(
mod tests {
use super::{insert, notify_new_transactions, update_expenditures, update_transactions};
use crate::ingestion::ingestion_logic::from_json_row;
use anyhow::Error;
use tokio::sync::OnceCell;
use entity::account;
use migration::MigratorTrait;
use sea_orm::{DatabaseConnection, TransactionTrait};
use sea_orm::{ActiveModelTrait, DatabaseConnection, TransactionTrait};
use serde_json::Value;
use sqlx::postgres::PgListener;
use sqlx::PgPool;
use sqlx::{Executor, PgPool};
use testcontainers::runners::AsyncRunner;
use testcontainers::ContainerAsync;
use crate::ingestion::ingestion_logic::from_json_row;
use tokio::sync::OnceCell;
#[derive(Debug)]
struct DatabaseInstance {
@ -203,13 +204,19 @@ mod tests {
Ok(instance)
}
async fn get_or_initialize_db_instance() -> Result<
&'static DatabaseInstance,
Error,
> {
Ok(INSTANCE.get_or_init(|| async {
initialise_db().await.unwrap()
}).await)
async fn get_or_initialize_db_instance() -> Result<&'static DatabaseInstance, Error> {
Ok(INSTANCE
.get_or_init(|| async { initialise_db().await.unwrap() })
.await)
}
async fn create_test_account(db: &DatabaseConnection) -> Result<i32, Error> {
let new_account = account::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
name: sea_orm::ActiveValue::Set("Test Account".to_string()),
};
let inserted = new_account.insert(db).await?;
Ok(inserted.id)
}
#[tokio::test]
@ -253,6 +260,7 @@ mod tests {
#[tokio::test]
async fn test_notify_on_insert() -> Result<(), Error> {
let dbi = get_or_initialize_db_instance().await?;
let account_id = create_test_account(&dbi.db).await?;
let mut listener = PgListener::connect_with(&dbi.pool).await?;
listener.listen("monzo_new_transactions").await?;
@ -261,16 +269,16 @@ mod tests {
let data = json
.iter()
.map(|row| from_json_row(row.clone()))
.collect::<Result<Vec<_>, anyhow::Error>>()
.unwrap();
.collect::<Result<Vec<_>, anyhow::Error>>()?;
insert(&dbi.db, data.clone(), ).await?;
insert(&dbi.db, data.clone(), account_id).await?;
let notification = listener.recv().await?;
let payload = notification.payload();
let mut payload = serde_json::from_str::<Vec<String>>(&payload)?;
payload.sort();
let mut ids = data.iter()
let mut ids = data
.iter()
.map(|row| row.transaction_id.clone())
.collect::<Vec<_>>();
@ -278,28 +286,43 @@ mod tests {
assert_eq!(payload, ids, "Inserted IDs do not match");
insert(&dbi.db, data.clone(), ).await?;
insert(&dbi.db, data.clone(), account_id).await?;
let notification = listener.recv().await?;
let payload = notification.payload();
let payload = serde_json::from_str::<Vec<String>>(&payload)?;
assert_eq!(payload, Vec::<String>::new(), "Re-inserting identical rows triggered double notification");
assert_eq!(
payload,
Vec::<String>::new(),
"Re-inserting identical rows triggered double notification"
);
let mut altered_data = data.clone();
altered_data[0].description = Some("New description".to_string());
assert_ne!(altered_data[0].compute_hash(), data[0].compute_hash(), "Alterations have the same hash");
assert_ne!(
altered_data[0].compute_hash(),
data[0].compute_hash(),
"Alterations have the same hash"
);
insert(&dbi.db, altered_data.clone(), ).await?;
insert(&dbi.db, altered_data.clone(), account_id).await?;
let notification = listener.recv().await?;
let payload = notification.payload();
let payload = serde_json::from_str::<Vec<String>>(&payload)?;
assert_eq!(payload, vec![altered_data[0].transaction_id.clone()], "Re-inserting altered row failed to re-trigger notification");
assert_eq!(
payload,
vec![altered_data[0].transaction_id.clone()],
"Re-inserting altered row failed to re-trigger notification"
);
Ok(())
}
}
pub(crate) async fn get_account_id(p0: &DatabaseConnection, p1: Option<String>) -> Result<i32, AppError> {
pub(crate) async fn get_account_id(
p0: &DatabaseConnection,
p1: Option<String>,
) -> Result<i32, AppError> {
let p1 = p1.unwrap_or("Monzo".to_string());
entity::prelude::Account::find()
@ -307,6 +330,7 @@ pub(crate) async fn get_account_id(p0: &DatabaseConnection, p1: Option<String>)
.select_only()
.column(entity::account::Column::Id)
.into_tuple::<i32>()
.one(p0).await?
.one(p0)
.await?
.ok_or(AppError::BadRequest(anyhow!("Account not found")))
}