testing omni gen (#1875)

This commit is contained in:
hanashi authored and GitHub committed 2026-08-09 22:54:25 -04:00
1 parent b36e4fd27b
commit c62a269638
50 files changed
+6676 -51

No files matched your search

+20
View File
@@ -1,5 +1,25 @@
# Testing
## Database Fixture Tests
- Live in-crate (e.g. `storyteller-web`'s omni_gen video pricing tests) and
use the `mysql_testing` crate (`crates/schema/database/mysql_testing`) for
guarded pools, schema setup, and account/session/wallet fixtures
- They RUN BY DEFAULT with `cargo test -p storyteller-web` and from the IDE
(requires a local MySQL with an `artcraft_test` database). Machines/CI
without MySQL skip them via the off switch:
`SQLX_OFFLINE=true cargo test -p storyteller-web --features skip_database_tests`
- They connect ONLY via `ARTCRAFT_TEST_DATABASE_URL` (default
`mysql://root:@localhost:3306/artcraft_test`); the guard panics on any
database whose name lacks "test", on `storyteller`/`artcraft`, and on
cloud hosts — see `mysql_testing/README.md`
- Database tests run in PARALLEL: each test creates its own users, wallets,
and rows; schema setup single-flights via a MySQL named lock. Only tests
mutating shared/global state take
`mysql_testing::serial::acquire_serial_test_lock().await`
- External providers must never be called: the harness stubs Kinovi with an
in-process HTTP server via `KINOVI_CUSTOM_API_HOST`/`KINOVI_CUSTOM_CDN_HOST`
## Running Tests
- Most crates: `cargo test -p {crate_name}`
Generated
+17
View File
@@ -7363,6 +7363,22 @@ dependencies = [
"workspace-build-acceleration",
]
[[package]]
name = "mysql_testing"
version = "0.1.0"
dependencies = [
"anyhow",
"bcrypt",
"bucket_paths",
"chrono",
"enums",
"log",
"mysql_queries",
"sqlx",
"tokens",
"tokio",
]
[[package]]
name = "mysqlclient-sys"
version = "0.2.5"
@@ -11646,6 +11662,7 @@ dependencies = [
"migration",
"mimetypes",
"mysql_queries",
"mysql_testing",
"once_cell",
"opaque_cursors",
"pager",
+2
View File
@@ -120,6 +120,7 @@ members = [
"crates/schema/database/elasticsearch_schema",
"crates/schema/database/migration",
"crates/schema/database/mysql_queries",
"crates/schema/database/mysql_testing",
"crates/schema/database/redis_common",
"crates/schema/database/redis_schema",
"crates/schema/database/sqlite_tasks",
@@ -282,6 +283,7 @@ mimetypes = { path = "crates/lib/files/mimetypes" }
muapi_client = { path = "crates/api_clients/muapi_client" }
metrics = { path = "crates/lib/metrics" }
mysql_queries = { path = "crates/schema/database/mysql_queries" }
mysql_testing = { path = "crates/schema/database/mysql_testing" }
pager = { path = "crates/lib/pager" }
openai_sora_client = { path = "crates/api_clients/openai_sora_client" }
primitives = { path = "crates/lib/data/primitives" }
@@ -255,6 +255,30 @@ pub fn plan_batch_count(
}
}
/// Batch counts for the Seedance 2.0 Mini family: 1-8 are all valid
/// (wider than the shared 1/2/4 planning); higher counts clamp to 8 per the
/// mitigation strategy.
pub fn plan_mini_batch_count(
video_batch_count: Option<u16>,
strategy: RequestMismatchMitigationStrategy,
) -> Result<u16, ArtcraftRouterError> {
let count = video_batch_count.unwrap_or(1);
match count {
0 => Err(ArtcraftRouterError::Client(ClientError::UserRequestedZeroGenerations)),
1..=8 => Ok(count),
_ => match strategy {
RequestMismatchMitigationStrategy::ErrorOut => {
Err(ArtcraftRouterError::Client(ClientError::ModelDoesNotSupportOption {
field: "video_batch_count",
value: format!("{}", count),
}))
}
RequestMismatchMitigationStrategy::PayMoreUpgrade
| RequestMismatchMitigationStrategy::PayLessDowngrade => Ok(8),
},
}
}
/// Translate the router-facing bitrate into the API `CommonBitrate`.
///
/// Bitrate does not affect cost. An unset value leaves the field `None` (the
@@ -3,21 +3,28 @@ use enums::common::generation::common_video_model::CommonVideoModel as CommonVid
use crate::errors::artcraft_router_error::ArtcraftRouterError;
use crate::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder;
use crate::generate::generate_video::providers::artcraft::build_common::{
build_artcraft_omni_video_request, SupportedResolutions, UltraWideSupport,
build_artcraft_omni_video_request, plan_mini_batch_count, SupportedResolutions, UltraWideSupport,
};
use crate::generate::generate_video::providers::artcraft::seedance_2p0_bp_mini::request::ArtcraftSeedance2p0BytePlusMiniRequestState;
use crate::generate::generate_video::video_generation_draft_or_request::VideoGenerationDraftOrRequest;
use crate::generate::generate_video::video_generation_request::VideoGenerationRequest;
pub fn build_artcraft_seedance_2p0_bp_mini(builder: GenerateVideoRequestBuilder) -> Result<VideoGenerationDraftOrRequest, ArtcraftRouterError> {
pub fn build_artcraft_seedance_2p0_bp_mini(mut builder: GenerateVideoRequestBuilder) -> Result<VideoGenerationDraftOrRequest, ArtcraftRouterError> {
// Seedance 2.0 Mini supports 480p and 720p only (higher resolutions downgrade
// to 720p), and all six aspect ratios including 21:9.
let request = build_artcraft_omni_video_request(
// Mini supports batches of 1-8, wider than the shared 1/2/4 planning.
// Take the batch out before the shared build and set it after.
let strategy = builder.request_mismatch_mitigation_strategy;
let batch_count = plan_mini_batch_count(builder.video_batch_count.take(), strategy)?;
let mut request = build_artcraft_omni_video_request(
builder,
CommonVideoModelEnum::Seedance2p0BytePlusMini,
SupportedResolutions::Fast,
UltraWideSupport::Supported,
)?;
request.video_batch_count = Some(batch_count);
let state = ArtcraftSeedance2p0BytePlusMiniRequestState { request };
Ok(VideoGenerationDraftOrRequest::Request(VideoGenerationRequest::ArtcraftSeedance2p0BytePlusMini(state)))
}
@@ -80,10 +87,18 @@ mod tests {
}
#[test]
fn batch_count_clamped_to_platform_max() {
// The ArtCraft platform caps batch at 4 and downgrades higher requests.
let req = unwrap_request(builder_with(|b| { b.video_batch_count = Some(8); }));
assert_eq!(req.request.video_batch_count, Some(4));
fn batch_count_up_to_eight_passed_through() {
// Mini supports batches of 1-8; execution and billing must agree.
for batch in 1..=8u16 {
let req = unwrap_request(builder_with(move |b| { b.video_batch_count = Some(batch); }));
assert_eq!(req.request.video_batch_count, Some(batch));
}
}
#[test]
fn batch_count_above_eight_clamps_to_eight() {
let req = unwrap_request(builder_with(|b| { b.video_batch_count = Some(9); }));
assert_eq!(req.request.video_batch_count, Some(8));
}
#[test]
@@ -105,6 +105,49 @@ mod tests {
}
}
mod batch_pricing {
use super::*;
/// Mini bills batches of 1-8 in full (the platform's 1/2/4 cap does not
/// apply). Once, batch 8 executed in full but billed as batch 4.
#[test]
fn every_batch_size_720p_5s() {
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 1, false), 44);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 2, false), 88);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 3, false), 131);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 4, false), 175);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 5, false), 219);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 6, false), 262);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 7, false), 306);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 8, false), 349);
}
#[test]
fn every_batch_size_480p_5s() {
assert_eq!(cents(RouterResolution::FourEightyP, 5, 1, false), 17);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 2, false), 33);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 3, false), 50);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 4, false), 66);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 5, false), 82);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 6, false), 99);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 7, false), 115);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 8, false), 131);
}
#[test]
fn batch_eight_with_video_reference() {
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 8, true), 419);
}
#[test]
fn batch_above_eight_clamps_to_eight() {
assert_eq!(
cents(RouterResolution::SevenTwentyP, 5, 9, false),
cents(RouterResolution::SevenTwentyP, 5, 8, false),
);
}
}
mod credits_tests {
use super::*;
@@ -3,21 +3,28 @@ use enums::common::generation::common_video_model::CommonVideoModel as CommonVid
use crate::errors::artcraft_router_error::ArtcraftRouterError;
use crate::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder;
use crate::generate::generate_video::providers::artcraft::build_common::{
build_artcraft_omni_video_request, SupportedResolutions, UltraWideSupport,
build_artcraft_omni_video_request, plan_mini_batch_count, SupportedResolutions, UltraWideSupport,
};
use crate::generate::generate_video::providers::artcraft::seedance_2p0_bpu_mini::request::ArtcraftSeedance2p0BytePlusUltraMiniRequestState;
use crate::generate::generate_video::video_generation_draft_or_request::VideoGenerationDraftOrRequest;
use crate::generate::generate_video::video_generation_request::VideoGenerationRequest;
pub fn build_artcraft_seedance_2p0_bpu_mini(builder: GenerateVideoRequestBuilder) -> Result<VideoGenerationDraftOrRequest, ArtcraftRouterError> {
pub fn build_artcraft_seedance_2p0_bpu_mini(mut builder: GenerateVideoRequestBuilder) -> Result<VideoGenerationDraftOrRequest, ArtcraftRouterError> {
// Seedance 2.0 Mini supports 480p and 720p only (higher resolutions downgrade
// to 720p), and all six aspect ratios including 21:9.
let request = build_artcraft_omni_video_request(
// Mini supports batches of 1-8, wider than the shared 1/2/4 planning.
// Take the batch out before the shared build and set it after.
let strategy = builder.request_mismatch_mitigation_strategy;
let batch_count = plan_mini_batch_count(builder.video_batch_count.take(), strategy)?;
let mut request = build_artcraft_omni_video_request(
builder,
CommonVideoModelEnum::Seedance2p0BytePlusUltraMini,
SupportedResolutions::Fast,
UltraWideSupport::Supported,
)?;
request.video_batch_count = Some(batch_count);
let state = ArtcraftSeedance2p0BytePlusUltraMiniRequestState { request };
Ok(VideoGenerationDraftOrRequest::Request(VideoGenerationRequest::ArtcraftSeedance2p0BytePlusUltraMini(state)))
}
@@ -80,10 +87,18 @@ mod tests {
}
#[test]
fn batch_count_clamped_to_platform_max() {
// The ArtCraft platform caps batch at 4 and downgrades higher requests.
let req = unwrap_request(builder_with(|b| { b.video_batch_count = Some(8); }));
assert_eq!(req.request.video_batch_count, Some(4));
fn batch_count_up_to_eight_passed_through() {
// Mini supports batches of 1-8; execution and billing must agree.
for batch in 1..=8u16 {
let req = unwrap_request(builder_with(move |b| { b.video_batch_count = Some(batch); }));
assert_eq!(req.request.video_batch_count, Some(batch));
}
}
#[test]
fn batch_count_above_eight_clamps_to_eight() {
let req = unwrap_request(builder_with(|b| { b.video_batch_count = Some(9); }));
assert_eq!(req.request.video_batch_count, Some(8));
}
#[test]
@@ -105,6 +105,49 @@ mod tests {
}
}
mod batch_pricing {
use super::*;
/// Mini bills batches of 1-8 in full (the platform's 1/2/4 cap does not
/// apply). Once, batch 8 executed in full but billed as batch 4.
#[test]
fn every_batch_size_720p_5s() {
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 1, false), 44);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 2, false), 88);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 3, false), 131);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 4, false), 175);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 5, false), 219);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 6, false), 262);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 7, false), 306);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 8, false), 349);
}
#[test]
fn every_batch_size_480p_5s() {
assert_eq!(cents(RouterResolution::FourEightyP, 5, 1, false), 17);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 2, false), 33);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 3, false), 50);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 4, false), 66);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 5, false), 82);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 6, false), 99);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 7, false), 115);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 8, false), 131);
}
#[test]
fn batch_eight_with_video_reference() {
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 8, true), 419);
}
#[test]
fn batch_above_eight_clamps_to_eight() {
assert_eq!(
cents(RouterResolution::SevenTwentyP, 5, 9, false),
cents(RouterResolution::SevenTwentyP, 5, 8, false),
);
}
}
mod credits_tests {
use super::*;
@@ -3,21 +3,28 @@ use enums::common::generation::common_video_model::CommonVideoModel as CommonVid
use crate::errors::artcraft_router_error::ArtcraftRouterError;
use crate::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder;
use crate::generate::generate_video::providers::artcraft::build_common::{
build_artcraft_omni_video_request, SupportedResolutions, UltraWideSupport,
build_artcraft_omni_video_request, plan_mini_batch_count, SupportedResolutions, UltraWideSupport,
};
use crate::generate::generate_video::providers::artcraft::seedance_2p0_mini::request::ArtcraftSeedance2p0MiniRequestState;
use crate::generate::generate_video::video_generation_draft_or_request::VideoGenerationDraftOrRequest;
use crate::generate::generate_video::video_generation_request::VideoGenerationRequest;
pub fn build_artcraft_seedance_2p0_mini(builder: GenerateVideoRequestBuilder) -> Result<VideoGenerationDraftOrRequest, ArtcraftRouterError> {
pub fn build_artcraft_seedance_2p0_mini(mut builder: GenerateVideoRequestBuilder) -> Result<VideoGenerationDraftOrRequest, ArtcraftRouterError> {
// Seedance 2.0 Mini supports 480p and 720p only (higher resolutions downgrade
// to 720p), and all six aspect ratios including 21:9.
let request = build_artcraft_omni_video_request(
// Mini supports batches of 1-8, wider than the shared 1/2/4 planning.
// Take the batch out before the shared build and set it after.
let strategy = builder.request_mismatch_mitigation_strategy;
let batch_count = plan_mini_batch_count(builder.video_batch_count.take(), strategy)?;
let mut request = build_artcraft_omni_video_request(
builder,
CommonVideoModelEnum::Seedance2p0Mini,
SupportedResolutions::Fast,
UltraWideSupport::Supported,
)?;
request.video_batch_count = Some(batch_count);
let state = ArtcraftSeedance2p0MiniRequestState { request };
Ok(VideoGenerationDraftOrRequest::Request(VideoGenerationRequest::ArtcraftSeedance2p0Mini(state)))
}
@@ -80,10 +87,18 @@ mod tests {
}
#[test]
fn batch_count_clamped_to_platform_max() {
// The ArtCraft platform caps batch at 4 and downgrades higher requests.
let req = unwrap_request(builder_with(|b| { b.video_batch_count = Some(8); }));
assert_eq!(req.request.video_batch_count, Some(4));
fn batch_count_up_to_eight_passed_through() {
// Mini supports batches of 1-8; execution and billing must agree.
for batch in 1..=8u16 {
let req = unwrap_request(builder_with(move |b| { b.video_batch_count = Some(batch); }));
assert_eq!(req.request.video_batch_count, Some(batch));
}
}
#[test]
fn batch_count_above_eight_clamps_to_eight() {
let req = unwrap_request(builder_with(|b| { b.video_batch_count = Some(9); }));
assert_eq!(req.request.video_batch_count, Some(8));
}
#[test]
@@ -105,6 +105,49 @@ mod tests {
}
}
mod batch_pricing {
use super::*;
/// Mini bills batches of 1-8 in full (the platform's 1/2/4 cap does not
/// apply). Once, batch 8 executed in full but billed as batch 4.
#[test]
fn every_batch_size_720p_5s() {
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 1, false), 44);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 2, false), 87);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 3, false), 130);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 4, false), 173);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 5, false), 217);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 6, false), 260);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 7, false), 303);
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 8, false), 346);
}
#[test]
fn every_batch_size_480p_5s() {
assert_eq!(cents(RouterResolution::FourEightyP, 5, 1, false), 17);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 2, false), 33);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 3, false), 49);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 4, false), 65);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 5, false), 82);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 6, false), 98);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 7, false), 114);
assert_eq!(cents(RouterResolution::FourEightyP, 5, 8, false), 130);
}
#[test]
fn batch_eight_with_video_reference() {
assert_eq!(cents(RouterResolution::SevenTwentyP, 5, 8, true), 415);
}
#[test]
fn batch_above_eight_clamps_to_eight() {
assert_eq!(
cents(RouterResolution::SevenTwentyP, 5, 9, false),
cents(RouterResolution::SevenTwentyP, 5, 8, false),
);
}
}
mod credits_tests {
use super::*;
@@ -1,5 +1,4 @@
use enums::common::generation::common_resolution::CommonResolution;
use kinovi_web_client::generate::video::generate_seedance_2p5::MAX_BILLED_INPUT_SECONDS;
use crate::generate::generate_video::video_generation_cost_estimate::VideoGenerationCostEstimate;
use crate::generate::generate_video::providers::artcraft::seedance_common::seedance_2p5_usd_cents;
@@ -8,9 +7,8 @@ use crate::generate::generate_video::providers::artcraft::seedance_2p5::request:
/// Seedance 2.5 pricing depends on the resolution, the output duration, and
/// — when reference videos are attached — the total seconds of reference
/// video input, which are billed on top of the output duration (at a lower
/// per-second rate). Input seconds are clamped to
/// [`MAX_BILLED_INPUT_SECONDS`] — the model accepts at most 30 seconds of
/// video. No batching.
/// per-second rate). The TOTAL input duration clamps to the 4..=30 second
/// billing range. No batching.
pub struct ArtcraftSeedance2p5CostState {
pub resolution: CommonResolution,
pub duration_seconds: u16,
@@ -37,8 +35,9 @@ impl ArtcraftSeedance2p5CostState {
}
pub fn estimate_cost(&self) -> VideoGenerationCostEstimate {
let total_input_seconds = self.total_input_seconds
.min(u16::from(MAX_BILLED_INPUT_SECONDS));
// Input-second clamping (the 4..=30 billing range) happens inside the
// shared pricing function.
let total_input_seconds = self.total_input_seconds;
let usd_cents = seedance_2p5_usd_cents(
self.resolution,
@@ -116,9 +115,19 @@ mod tests {
}
#[test]
fn missing_input_seconds_bill_output_duration_only() {
// 7.24279835 ¢/s × 10 = 72.43 → 73¢.
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, None), 73);
fn missing_input_seconds_bill_the_four_second_minimum() {
// No measured input still bills the 4-second total minimum:
// 7.24279835 ¢/s × (10+4) = 101.40 → 102¢.
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, None), 102);
}
#[test]
fn input_totals_under_four_seconds_clamp_to_four() {
// 1s, 3s, and 4s totals price identically; 5s bills more.
let at_minimum = cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(4));
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(1)), at_minimum);
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(3)), at_minimum);
assert!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(5)) > at_minimum);
}
#[test]
@@ -1,5 +1,4 @@
use enums::common::generation::common_resolution::CommonResolution;
use kinovi_web_client::generate::video::generate_seedance_2p5::MAX_BILLED_INPUT_SECONDS;
use crate::generate::generate_video::video_generation_cost_estimate::VideoGenerationCostEstimate;
use crate::generate::generate_video::providers::artcraft::seedance_common::seedance_2p5_ultra_usd_cents;
@@ -37,8 +36,9 @@ impl ArtcraftSeedance2p5UltraCostState {
}
pub fn estimate_cost(&self) -> VideoGenerationCostEstimate {
let total_input_seconds = self.total_input_seconds
.min(u16::from(MAX_BILLED_INPUT_SECONDS));
// Input-second clamping (the 4..=30 billing range) happens inside the
// shared pricing function.
let total_input_seconds = self.total_input_seconds;
let usd_cents = seedance_2p5_ultra_usd_cents(
self.resolution,
@@ -133,9 +133,18 @@ mod tests {
}
#[test]
fn missing_input_seconds_bill_output_duration_only() {
// 8.55967078 ¢/s × 10 = 85.60 → 86¢.
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, None), 86);
fn missing_input_seconds_bill_the_four_second_minimum() {
// No measured input still bills the 4-second total minimum:
// 8.55967078 ¢/s × (10+4) = 119.84 → 120¢.
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, None), 120);
}
#[test]
fn input_totals_under_four_seconds_clamp_to_four() {
let at_minimum = cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(4));
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(1)), at_minimum);
assert_eq!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(3)), at_minimum);
assert!(cents_with_video_refs(Some(RouterResolution::FourEightyP), 10, Some(5)) > at_minimum);
}
#[test]
@@ -3,6 +3,9 @@
//! ArtCraft credits equal USD cents (100 credits = $1.00).
use enums::common::generation::common_resolution::CommonResolution;
use kinovi_web_client::generate::video::generate_seedance_2p5::{
MAX_BILLED_INPUT_SECONDS, MIN_BILLED_INPUT_SECONDS,
};
// 4K is priced uniformly across the non-Fast Seedance 2.0 models. Rates are held
// in hundredths of a USD cent per second so the math is exact integer arithmetic
@@ -155,8 +158,9 @@ const SEEDANCE_2P5_VIDEO_REFERENCE_CENTS_PER_SECOND_720P: f64 = 15.84362140;
/// Only 480p and 720p are offered; any other resolution prices at 720p.
/// Without video references, billed seconds = output duration. With video
/// references, the per-second rate drops but the billed seconds are the
/// output duration PLUS the total seconds of reference video input. The
/// fractional total is rounded UP to a whole cent. No batching.
/// output duration PLUS the total seconds of reference video input (clamped
/// to the 4..=30 second billing range). The fractional total is rounded UP
/// to a whole cent. No batching.
pub fn seedance_2p5_usd_cents(
resolution: CommonResolution,
duration_seconds: u16,
@@ -169,7 +173,11 @@ pub fn seedance_2p5_usd_cents(
// Everything else (including 720p and unsupported resolutions) prices at 720p.
_ => SEEDANCE_2P5_VIDEO_REFERENCE_CENTS_PER_SECOND_720P,
};
(rate, u64::from(duration_seconds) + u64::from(total_input_seconds))
// The TOTAL input duration clamps to the 4..=30 second billing range
// (three 1s videos sum to 3 and bill 4; three 3s videos bill 9).
let billed_input_seconds = total_input_seconds
.clamp(u16::from(MIN_BILLED_INPUT_SECONDS), u16::from(MAX_BILLED_INPUT_SECONDS));
(rate, u64::from(duration_seconds) + u64::from(billed_input_seconds))
} else {
let rate = match resolution {
CommonResolution::FourEightyP => SEEDANCE_2P5_CENTS_PER_SECOND_480P,
@@ -209,7 +217,11 @@ pub fn seedance_2p5_ultra_usd_cents(
// Everything else (including 720p and unsupported resolutions) prices at 720p.
_ => SEEDANCE_2P5_ULTRA_VIDEO_REFERENCE_CENTS_PER_SECOND_720P,
};
(rate, u64::from(duration_seconds) + u64::from(total_input_seconds))
// The TOTAL input duration clamps to the 4..=30 second billing range
// (three 1s videos sum to 3 and bill 4; three 3s videos bill 9).
let billed_input_seconds = total_input_seconds
.clamp(u16::from(MIN_BILLED_INPUT_SECONDS), u16::from(MAX_BILLED_INPUT_SECONDS));
(rate, u64::from(duration_seconds) + u64::from(billed_input_seconds))
} else {
let rate = match resolution {
CommonResolution::FourEightyP => SEEDANCE_2P5_ULTRA_CENTS_PER_SECOND_480P,
@@ -1,6 +1,6 @@
use kinovi_web_client::generate::video::generate_seedance_2p5::{
GenerateSeedance2p5Request, KinoviSeedance2p5Modality, KinoviSeedance2p5OutputResolution,
MAX_BILLED_INPUT_SECONDS,
MAX_BILLED_INPUT_SECONDS, MIN_BILLED_INPUT_SECONDS,
};
use crate::api::video_list_ref::VideoListRef;
@@ -76,7 +76,7 @@ impl KinoviSeedance2p5CostState {
// NB: The kinovi calculator clamps too; clamping here as well keeps this
// state's math correct even if it's read directly.
let total_input_seconds = self.total_input_seconds
.map(|seconds| seconds.min(MAX_BILLED_INPUT_SECONDS));
.map(|seconds| seconds.clamp(MIN_BILLED_INPUT_SECONDS, MAX_BILLED_INPUT_SECONDS));
let pricing_request = GenerateSeedance2p5Request {
prompt: String::new(),
@@ -13,6 +13,12 @@ use crate::requests::workflow_run_task::workflow_run_task::{
/// seconds beyond this are clamped for billing.
pub const MAX_BILLED_INPUT_SECONDS: u8 = 30;
/// The minimum TOTAL billed input seconds when video references are
/// attached: however short the input videos are, the total bills at least 4
/// seconds. (Three 1-second videos sum to 3 and clamp to 4; three 3-second
/// videos sum to 9 and bill 9.)
pub const MIN_BILLED_INPUT_SECONDS: u8 = 4;
// ── Args ──
pub struct GenerateSeedance2p5Args<'a> {
@@ -118,7 +124,7 @@ pub enum KinoviSeedance2p5OutputResolution {
// With video references, the rate drops but the billed seconds are the
// output duration PLUS the total seconds of reference video input
// (`total_input_seconds`, summed across all reference videos and clamped to
// [`MAX_BILLED_INPUT_SECONDS`]). E.g. a 30s output with a 10s video
// the [`MIN_BILLED_INPUT_SECONDS`]..=[`MAX_BILLED_INPUT_SECONDS`] range). E.g. a 30s output with a 10s video
// reference bills as 40 seconds; a 30s output with two 7s references (14s
// total input) bills as 44 seconds.
//
@@ -141,7 +147,7 @@ impl GenerateSeedance2p5Request {
};
let input_seconds = self.total_input_seconds
.unwrap_or(MAX_BILLED_INPUT_SECONDS)
.min(MAX_BILLED_INPUT_SECONDS);
.clamp(MIN_BILLED_INPUT_SECONDS, MAX_BILLED_INPUT_SECONDS);
let seconds = u64::from(self.duration_seconds) + u64::from(input_seconds);
(rate, seconds)
} else {
@@ -392,6 +398,16 @@ mod tests {
assert_eq!(video_ref_720(30, Some(10)).calculate_costs().kinovi_credits, (35 * 40) as f64);
}
#[test]
fn input_totals_under_four_seconds_clamp_to_four() {
// Total input clamps to a 4-second minimum: 1s, 3s, and 4s totals
// all bill 4; 5s bills 5.
assert_eq!(video_ref_480(30, Some(1)).calculate_costs().kinovi_credits, (16 * 34) as f64);
assert_eq!(video_ref_480(30, Some(3)).calculate_costs().kinovi_credits, (16 * 34) as f64);
assert_eq!(video_ref_480(30, Some(4)).calculate_costs().kinovi_credits, (16 * 34) as f64);
assert_eq!(video_ref_480(30, Some(5)).calculate_costs().kinovi_credits, (16 * 35) as f64);
}
#[test]
fn missing_total_input_seconds_bills_the_worst_case_maximum() {
// Unknown input duration with video references attached: assume the
@@ -46,9 +46,30 @@ impl KinoviHost {
}
}
/// Env var pair for a process-wide custom host (both must be set and
/// non-empty to take effect). Intended for tests and local development
/// against a stub server; leave unset in production.
pub const ENV_KINOVI_CUSTOM_API_HOST: &str = "KINOVI_CUSTOM_API_HOST";
pub const ENV_KINOVI_CUSTOM_CDN_HOST: &str = "KINOVI_CUSTOM_CDN_HOST";
/// Resolves an optional host override to the effective host.
pub fn resolve_host(host_override: Option<&KinoviHost>) -> &KinoviHost {
// Use a static default to avoid needing to return owned data
static DEFAULT: KinoviHost = KinoviHost::Kinovi;
host_override.unwrap_or(&DEFAULT)
///
/// Precedence: explicit `host_override` argument, then the
/// [`ENV_KINOVI_CUSTOM_API_HOST`] / [`ENV_KINOVI_CUSTOM_CDN_HOST`] env var
/// pair, then the default host.
pub fn resolve_host(host_override: Option<&KinoviHost>) -> KinoviHost {
if let Some(host) = host_override {
return host.clone();
}
if let (Ok(api_host), Ok(cdn_host)) = (
std::env::var(ENV_KINOVI_CUSTOM_API_HOST),
std::env::var(ENV_KINOVI_CUSTOM_CDN_HOST),
) {
if !api_host.is_empty() && !cdn_host.is_empty() {
return KinoviHost::CustomHost { api_host, cdn_host };
}
}
KinoviHost::default()
}
@@ -80,7 +80,7 @@ pub async fn upload_file(args: UploadFileArgs) -> Result<UploadFileResponse, Kin
}.into());
}
let public_url = build_public_url(&args.upload_url, host)?;
let public_url = build_public_url(&args.upload_url, &host)?;
info!("Public URL: {}", public_url);
@@ -0,0 +1,26 @@
[package]
name = "mysql_testing"
version = "0.1.0"
edition = "2021"
publish = false
# Test-only database harness: guarded connections to a MySQL TEST database,
# schema setup, and shared fixtures (accounts, sessions, wallets).
#
# This crate must only ever be used as a dev-dependency.
[dependencies]
anyhow = "1.0"
bcrypt = "0.15"
bucket_paths = { path = "../../buckets/bucket_paths" }
chrono = { version = "0.4", features = ["serde"] }
log = "0.4"
sqlx = { version = "0.7.4", features = [ "mysql", "runtime-tokio-rustls", "chrono", "bigdecimal" ] }
tokio = { version = "1.50.0", features = ["macros"] }
enums = { path = "../../public/enums" }
mysql_queries = { path = "../mysql_queries" }
tokens = { path = "../../public/tokens" }
[dev-dependencies]
tokio = { version = "1.50.0", features = ["macros", "rt-multi-thread"] }
@@ -0,0 +1,67 @@
# mysql_testing
Test-only MySQL harness: guarded connections to a **test** database, schema
setup from the checked-in migrations, and shared fixtures (accounts,
sessions, wallets, media files). Use it as a `dev-dependency` only.
## Safety model
Every connection goes through `guard::guard_test_database_url`, which PANICS
unless:
- the database name contains `"test"`,
- the database name is not `storyteller` or `artcraft` (the real databases),
- the host does not look like a managed/cloud database (`digitalocean`,
`aws`, `amazonaws`, `rds`, `gcp`, `googleapis`, `cloudsql`, `azure`, ...).
Test code reads its URL from exactly one env var —
`ARTCRAFT_TEST_DATABASE_URL` — never `MYSQL_URL` or `DATABASE_URL`, so test
runs can't inherit a real database by accident. The default is
`mysql://root:@localhost:3306/artcraft_test`.
## One-time local setup
```bash
mysql -u root -e "CREATE DATABASE IF NOT EXISTS artcraft_test;"
```
The harness creates the database itself when the connecting user has the
privilege, applies all `_database/sql/migrations/*/up.sql` in order (tracked
in `mysql_testing_applied_migrations`, so repeat runs are fast), and seeds
`user_roles`. If the schema ever wedges: `DROP DATABASE artcraft_test;` and
rerun.
## Writing a database test
Database tests run by default; mark them with the `skip_database_tests`
off switch so database-less machines/CI can exclude them. Tests run in
PARALLEL — create your own users/wallets/rows and never touch another
test's data (schema setup single-flights via a MySQL named lock). Only a
test that must mutate shared/global state takes
`mysql_testing::serial::acquire_serial_test_lock()`.
```rust
#[tokio::test]
#[cfg_attr(feature = "skip_database_tests", ignore)]
async fn my_database_test() {
let pool = mysql_testing::pool::create_test_pool().await;
let user = mysql_testing::fixtures::users::create_test_user(&pool).await.unwrap();
mysql_testing::fixtures::wallets::fund_wallet_banked(&pool, &user.user_token, 10_000)
.await
.unwrap();
// ...
}
```
Skip them (machines/CI without MySQL) with:
```bash
SQLX_OFFLINE=true cargo test -p storyteller-web --features skip_database_tests
```
The first end-to-end consumers are the omni_gen video pricing tests in
`crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/tests/`,
which drive the real generate handler with dummy Actix requests, stub the
provider with an in-process HTTP server, and assert the exact credits
debited from the wallet.
@@ -0,0 +1,79 @@
//! Media file fixtures: rows a generation request can reference. The bucket
//! paths are fake — nothing downloads successfully from them, which tests
//! rely on (e.g. the reference-video probe falls back to worst-case billing).
use anyhow::anyhow;
use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath;
use enums::by_table::media_files::media_file_class::MediaFileClass;
use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory;
use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory;
use enums::by_table::media_files::media_file_type::MediaFileType;
use enums::common::visibility::Visibility;
use sqlx::MySqlPool;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::users::UserToken;
use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder;
/// Insert a video media_files row whose CDN path encodes `duration_millis`
/// as a `dur{millis}ms` prefix. Paired with a media-CDN override pointing at
/// a stub that serves real generated video of that duration, this lets tests
/// exercise the download + ffprobe billing path with exact durations. The
/// STORED duration column is deliberately wrong (999_000 ms) to prove the
/// database is never trusted for billing.
pub async fn create_probeable_test_video_media_file(
pool: &MySqlPool,
owner: &UserToken,
duration_millis: u64,
) -> anyhow::Result<MediaFileToken> {
let prefix = format!("dur{duration_millis}ms");
let bucket_path = MediaFileBucketPath::generate_new(Some(&prefix), Some("mp4"));
MediaFileInsertBuilder::new()
.creator_user(owner)
.creator_ip_address("127.0.0.1")
.creator_set_visibility(Visibility::Private)
.media_file_class(MediaFileClass::Video)
.media_file_type(MediaFileType::Video)
.is_user_upload(true)
.is_intermediate_system_file(false)
.media_file_origin_category(MediaFileOriginCategory::Upload)
.media_file_origin_product_category(MediaFileOriginProductCategory::Unknown)
.mime_type("video/mp4")
.file_size_bytes(1_024)
.maybe_duration_millis(Some(999_000))
.checksum_sha2(&"0".repeat(64))
.public_bucket_directory_hash(&bucket_path)
.insert_pool(pool)
.await
.map_err(|err| anyhow!("media file insert failed: {err:?}"))
}
/// Insert a video media_files row owned by `owner`, with the given stored
/// duration. Returns its token for use in `reference_video_media_tokens`.
pub async fn create_test_video_media_file(
pool: &MySqlPool,
owner: &UserToken,
maybe_duration_millis: Option<u64>,
) -> anyhow::Result<MediaFileToken> {
let bucket_path = MediaFileBucketPath::generate_new(None, Some("mp4"));
MediaFileInsertBuilder::new()
.creator_user(owner)
.creator_ip_address("127.0.0.1")
.creator_set_visibility(Visibility::Private)
.media_file_class(MediaFileClass::Video)
.media_file_type(MediaFileType::Video)
.is_user_upload(true)
.is_intermediate_system_file(false)
.media_file_origin_category(MediaFileOriginCategory::Upload)
.media_file_origin_product_category(MediaFileOriginProductCategory::Unknown)
.mime_type("video/mp4")
.file_size_bytes(1_024)
.maybe_duration_millis(maybe_duration_millis)
.checksum_sha2(&"0".repeat(64))
.public_bucket_directory_hash(&bucket_path)
.insert_pool(pool)
.await
.map_err(|err| anyhow!("media file insert failed: {err:?}"))
}
@@ -0,0 +1,3 @@
pub mod media_files;
pub mod users;
pub mod wallets;
@@ -0,0 +1,89 @@
//! Test accounts: a real `users` row (via the production create-account
//! query) plus a live `user_sessions` row, so handlers can authenticate the
//! fixture user exactly like a real one.
use std::sync::OnceLock;
use anyhow::anyhow;
use sqlx::MySqlPool;
use tokens::tokens::user_sessions::UserSessionToken;
use tokens::tokens::users::UserToken;
use mysql_queries::queries::users::user::create::create_account_from_email_and_password::{
create_account_from_email_and_password, CreateAccountFromEmailPasswordArgs,
};
use mysql_queries::queries::users::user_sessions::create_user_session_with_executor::create_user_session_with_executor;
/// Every fixture account uses this password (hashed once per process).
pub const TEST_USER_PASSWORD: &str = "test_password";
const TEST_IP_ADDRESS: &str = "127.0.0.1";
pub struct TestUser {
pub user_token: UserToken,
pub session_token: UserSessionToken,
pub username: String,
pub email_address: String,
}
/// Create a user with a unique username and an active session row.
pub async fn create_test_user(pool: &MySqlPool) -> anyhow::Result<TestUser> {
// Known a priori so the fixture can hand it back; the create-account query
// documents this hook as being for db integration tests and seeding.
// NB: the REAL generator, deliberately — the deterministic "for testing"
// generator repeats the same tokens in every process, which collides with
// rows left by previous test runs.
let user_token = UserToken::generate();
let suffix: String = user_token
.as_str()
.chars()
.rev()
.take(10)
.collect::<String>()
.to_lowercase();
let username = format!("test_{suffix}");
let email_address = format!("{username}@example-test.invalid");
let mut connection = pool.acquire().await?;
create_account_from_email_and_password(
CreateAccountFromEmailPasswordArgs {
username: &username,
display_name: &username,
email_address: &email_address,
email_gravatar_hash: "00000000000000000000000000000000",
password_hash: test_password_bcrypt_hash(),
ip_address: TEST_IP_ADDRESS,
maybe_source: None,
maybe_referral_url: None,
maybe_landing_url: None,
maybe_referral_partner: None,
maybe_referral_user_token: None,
maybe_user_token: Some(&user_token),
},
&mut connection,
)
.await
.map_err(|err| anyhow!("create_account_from_email_and_password failed: {err:?}"))?;
let session_token =
create_user_session_with_executor(&user_token, TEST_IP_ADDRESS, &mut *connection).await?;
Ok(TestUser {
user_token,
session_token,
username,
email_address,
})
}
/// Bcrypt hash of [`TEST_USER_PASSWORD`] at minimum cost, computed once.
/// (Tests authenticate via session rows, not passwords, but the column
/// requires a well-formed 60-byte bcrypt hash.)
fn test_password_bcrypt_hash() -> &'static str {
static HASH: OnceLock<String> = OnceLock::new();
HASH.get_or_init(|| {
bcrypt::hash(TEST_USER_PASSWORD, 4).expect("bcrypt hash")
})
}
@@ -0,0 +1,115 @@
//! Wallet fixtures: create/fund wallets and read back balances + ledger
//! entries for assertions.
use anyhow::anyhow;
use enums::common::payments_namespace::PaymentsNamespace;
use sqlx::{MySqlPool, Row};
use tokens::tokens::users::UserToken;
use tokens::tokens::wallets::WalletToken;
use mysql_queries::queries::wallets::add_durable_banked_balance_to_wallet::add_durable_banked_balance_to_wallet;
use mysql_queries::queries::wallets::create_new_artcraft_wallet_for_owner_user::create_new_artcraft_wallet_for_owner_user;
use mysql_queries::queries::wallets::find_primary_wallet_token_for_owner::find_primary_wallet_token_for_owner_using_transaction;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct WalletBalance {
pub banked_credits: u64,
pub monthly_credits: u64,
}
impl WalletBalance {
pub fn total(&self) -> u64 {
self.banked_credits + self.monthly_credits
}
}
#[derive(Clone, Debug)]
pub struct LedgerEntry {
pub entry_type: String,
pub credits_delta: i64,
pub is_refunded: bool,
pub maybe_entity_ref: Option<String>,
}
/// Ensure the user has an Artcraft wallet and add `credits` to its banked
/// balance. Returns the wallet token.
pub async fn fund_wallet_banked(
pool: &MySqlPool,
user_token: &UserToken,
credits: u64,
) -> anyhow::Result<WalletToken> {
let mut connection = pool.acquire().await?;
let mut transaction = sqlx::Acquire::begin(&mut connection).await?;
let maybe_wallet_token = find_primary_wallet_token_for_owner_using_transaction(
user_token,
PaymentsNamespace::Artcraft,
&mut transaction,
)
.await
.map_err(|err| anyhow!("find wallet failed: {err:?}"))?;
let wallet_token = match maybe_wallet_token {
Some(token) => token,
None => create_new_artcraft_wallet_for_owner_user(user_token, &mut transaction).await?,
};
if credits > 0 {
add_durable_banked_balance_to_wallet(
&wallet_token,
credits,
Some("mysql_testing fixture"),
None,
&mut transaction,
)
.await?;
}
transaction.commit().await?;
Ok(wallet_token)
}
/// The user's Artcraft wallet balance, or None if no wallet exists yet.
pub async fn artcraft_wallet_balance(
pool: &MySqlPool,
user_token: &UserToken,
) -> anyhow::Result<Option<WalletBalance>> {
let maybe_row = sqlx::query(
"SELECT banked_credits, monthly_credits FROM wallets \
WHERE owner_user_token = ? AND wallet_namespace = ? \
ORDER BY id ASC LIMIT 1",
)
.bind(user_token.as_str())
.bind(PaymentsNamespace::Artcraft.to_str())
.fetch_optional(pool)
.await?;
Ok(maybe_row.map(|row| WalletBalance {
banked_credits: u64::from(row.get::<u32, _>("banked_credits")),
monthly_credits: u64::from(row.get::<u32, _>("monthly_credits")),
}))
}
/// All ledger entries for the wallet, oldest first.
pub async fn wallet_ledger_entries(
pool: &MySqlPool,
wallet_token: &WalletToken,
) -> anyhow::Result<Vec<LedgerEntry>> {
let rows = sqlx::query(
"SELECT entry_type, credits_delta, is_refunded, maybe_entity_ref \
FROM wallet_ledger_entries WHERE wallet_token = ? ORDER BY id ASC",
)
.bind(wallet_token.as_str())
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| LedgerEntry {
entry_type: row.get::<String, _>("entry_type"),
credits_delta: i64::from(row.get::<i32, _>("credits_delta")),
is_refunded: row.get::<bool, _>("is_refunded"),
maybe_entity_ref: row.get::<Option<String>, _>("maybe_entity_ref"),
})
.collect())
}
@@ -0,0 +1,179 @@
//! The safety valve: refuse to touch anything that could be a real database.
//!
//! Every connection made by this crate goes through [`guard_test_database_url`]
//! first. It PANICS unless the URL names a database that is unambiguously a
//! test database on a non-production host. This makes it impossible for a
//! fixture test to wipe or bill against development or production data.
/// The ONE env var test code may read for a database URL.
/// Deliberately not `DATABASE_URL` (sqlx) or `MYSQL_CONNECTION_STRING` (the
/// server) so that test runs never inherit a real database by accident.
pub const TEST_DATABASE_URL_ENV: &str = "ARTCRAFT_TEST_DATABASE_URL";
/// Used when [`TEST_DATABASE_URL_ENV`] is unset. Still guarded.
pub const DEFAULT_TEST_DATABASE_URL: &str = "mysql://root:@localhost:3306/artcraft_test";
/// Database names that must NEVER be used, even if they somehow contained
/// "test": the production/development database names.
const FORBIDDEN_DATABASE_NAMES: [&str; 2] = ["storyteller", "artcraft"];
/// Host substrings that indicate a managed/cloud database. A test database
/// must never live on one of these.
const FORBIDDEN_HOST_SUBSTRINGS: [&str; 10] = [
"digitalocean",
"aws",
"amazonaws",
"rds",
"gcp",
"googleapis",
"cloudsql",
"azure",
"planetscale",
"supabase",
];
/// Panic unless `url` is a MySQL URL that names a test database on a
/// non-cloud host. Returns the parsed (host, database name) on success so
/// callers can log where they're connecting.
pub fn guard_test_database_url(url: &str) -> (String, String) {
let (host, database) = parse_mysql_url(url).unwrap_or_else(|| {
panic!("REFUSING to connect: could not parse a host and database name out of the test database URL");
});
let host_lower = host.to_lowercase();
let database_lower = database.to_lowercase();
for forbidden in FORBIDDEN_DATABASE_NAMES {
assert!(
database_lower != forbidden,
"REFUSING to connect: {database:?} is a real database name, not a test database. \
Tests must use a database whose name contains \"test\" (via {TEST_DATABASE_URL_ENV})."
);
}
assert!(
database_lower.contains("test"),
"REFUSING to connect: test database name must contain \"test\" (got {database:?}). \
Set {TEST_DATABASE_URL_ENV} to e.g. {DEFAULT_TEST_DATABASE_URL:?}."
);
for forbidden in FORBIDDEN_HOST_SUBSTRINGS {
assert!(
!host_lower.contains(forbidden),
"REFUSING to connect: host {host:?} looks like a managed/cloud database \
(matched {forbidden:?}). Tests must run against a local or dedicated test host."
);
}
(host, database)
}
/// Extract (host, database) from a `mysql://user:pass@host:port/database?...` URL.
fn parse_mysql_url(url: &str) -> Option<(String, String)> {
let rest = url.strip_prefix("mysql://")?;
let rest = rest.split('?').next()?;
// Strip credentials: everything up to the LAST '@' before the first '/'.
let (authority, database) = rest.split_once('/')?;
let host_and_port = match authority.rsplit_once('@') {
Some((_credentials, host_and_port)) => host_and_port,
None => authority,
};
// Strip the port. IPv6 hosts are bracketed: mysql://u@[::1]:3306/db
let host = if let Some(bracketed) = host_and_port.strip_prefix('[') {
bracketed.split_once(']').map(|(h, _)| h)?
} else {
host_and_port.split(':').next()?
};
if host.is_empty() || database.is_empty() {
return None;
}
Some((host.to_string(), database.to_string()))
}
#[cfg(test)]
mod tests {
use super::*;
fn guard_panics(url: &str) -> bool {
std::panic::catch_unwind(|| guard_test_database_url(url)).is_err()
}
mod accepted_urls {
use super::*;
#[test]
fn local_test_databases() {
for url in [
"mysql://root:@localhost:3306/artcraft_test",
"mysql://root:@127.0.0.1:3306/storyteller_test",
"mysql://user:pass@localhost/test_fixtures",
"mysql://user:pass@[::1]:3306/artcraft_test",
"mysql://ci:ci@mysql:3306/artcraft_test", // docker service host
] {
let (_host, database) = guard_test_database_url(url);
assert!(database.to_lowercase().contains("test"));
}
}
}
mod rejected_urls {
use super::*;
#[test]
fn production_and_development_database_names() {
assert!(guard_panics("mysql://root:@localhost:3306/storyteller"));
assert!(guard_panics("mysql://root:@localhost:3306/artcraft"));
assert!(guard_panics("mysql://root:@localhost:3306/Storyteller"));
}
#[test]
fn names_without_test() {
assert!(guard_panics("mysql://root:@localhost:3306/storyteller_dev"));
assert!(guard_panics("mysql://root:@localhost:3306/scratch"));
}
#[test]
fn cloud_hosts_even_with_test_names() {
assert!(guard_panics(
"mysql://doadmin:x@mysql-do-user-1-0.d.db.ondigitalocean.com:20000/artcraft_test"
));
assert!(guard_panics("mysql://u:p@db.abc123.us-east-1.rds.amazonaws.com/artcraft_test"));
assert!(guard_panics("mysql://u:p@some-gcp-host.googleapis.com/artcraft_test"));
assert!(guard_panics("mysql://u:p@my-db.azure.example/artcraft_test"));
}
#[test]
fn unparseable_urls() {
assert!(guard_panics("postgres://root:@localhost/artcraft_test"));
assert!(guard_panics("mysql://localhost"));
assert!(guard_panics(""));
}
}
mod url_parsing {
use super::*;
#[test]
fn parses_host_and_database() {
assert_eq!(
parse_mysql_url("mysql://u:p@localhost:3306/artcraft_test?ssl-mode=DISABLED"),
Some(("localhost".to_string(), "artcraft_test".to_string())),
);
}
#[test]
fn parses_password_containing_at_sign() {
assert_eq!(
parse_mysql_url("mysql://u:p%40ss@h.example:3306/artcraft_test"),
Some(("h.example".to_string(), "artcraft_test".to_string())),
);
assert_eq!(
parse_mysql_url("mysql://u:p@ss@h.example:3306/artcraft_test"),
Some(("h.example".to_string(), "artcraft_test".to_string())),
);
}
}
}
@@ -0,0 +1,27 @@
//! Test-only MySQL harness: guarded connections to a TEST database, schema
//! setup from the checked-in migrations, and shared fixtures (accounts,
//! sessions, wallets, media files).
//!
//! # Safety model
//!
//! Every connection is vetted by [`guard::guard_test_database_url`], which
//! panics unless the database name contains "test" (and is not a real
//! database name) and the host is not a managed/cloud database. See that
//! module for the exact rules. Tests read their URL from
//! [`guard::TEST_DATABASE_URL_ENV`] only — never `MYSQL_URL` or
//! `DATABASE_URL`.
//!
//! # Usage
//!
//! ```ignore
//! let _serial = mysql_testing::serial::acquire_serial_test_lock().await;
//! let pool = mysql_testing::pool::create_test_pool().await;
//! let user = mysql_testing::fixtures::users::create_test_user(&pool).await.unwrap();
//! mysql_testing::fixtures::wallets::fund_wallet_banked(&pool, &user.user_token, 10_000).await.unwrap();
//! ```
pub mod fixtures;
pub mod guard;
pub mod pool;
pub mod schema;
pub mod serial;
@@ -0,0 +1,111 @@
//! Guarded test-database connections.
use log::info;
use sqlx::mysql::MySqlPoolOptions;
use sqlx::MySqlPool;
use crate::guard::{guard_test_database_url, DEFAULT_TEST_DATABASE_URL, TEST_DATABASE_URL_ENV};
use crate::schema::ensure_schema;
/// Connect to the test database, creating it and applying the schema first.
///
/// Reads [`TEST_DATABASE_URL_ENV`]; falls back to
/// [`DEFAULT_TEST_DATABASE_URL`]. Panics (via the guard) if the URL points
/// anywhere that isn't unambiguously a test database.
pub async fn create_test_pool() -> MySqlPool {
let url = std::env::var(TEST_DATABASE_URL_ENV)
.unwrap_or_else(|_| DEFAULT_TEST_DATABASE_URL.to_string());
let (host, database) = guard_test_database_url(&url);
info!("Connecting to test database {database:?} on {host:?}");
// Local test databases don't need TLS, and test binaries often have
// ambiguous rustls crypto providers (which panics when sqlx negotiates
// TLS). Disable it unless the URL explicitly configures ssl-mode.
let url = if url.contains("ssl-mode") {
url
} else if url.contains('?') {
format!("{url}&ssl-mode=DISABLED")
} else {
format!("{url}?ssl-mode=DISABLED")
};
create_database_if_missing(&url, &database).await;
let pool = MySqlPoolOptions::new()
.max_connections(4)
.connect(&url)
.await
.unwrap_or_else(|err| {
panic!(
"Failed to connect to the test database ({url_env}={url:?}): {err}. \
Is MySQL running, and does the user have access?",
url_env = TEST_DATABASE_URL_ENV,
)
});
ensure_schema(&pool).await;
pool
}
/// `CREATE DATABASE IF NOT EXISTS` via a server-level (no database) connection.
/// Best-effort: if the user lacks the privilege but the database already
/// exists, the later pool connection succeeds anyway.
async fn create_database_if_missing(url: &str, database: &str) {
// The database name came out of the guard, but re-validate before splicing
// it into DDL (identifiers can't be bound as parameters).
assert!(
database.chars().all(|c| c.is_ascii_alphanumeric() || c == '_'),
"test database name must be [A-Za-z0-9_]+ (got {database:?})",
);
let server_url = strip_database_from_url(url);
let server_connection = MySqlPoolOptions::new()
.max_connections(1)
.connect(&server_url)
.await;
match server_connection {
Ok(pool) => {
let create = format!("CREATE DATABASE IF NOT EXISTS `{database}`");
if let Err(err) = sqlx::query(&create).execute(&pool).await {
info!("Could not CREATE DATABASE {database:?} ({err}); assuming it already exists");
}
pool.close().await;
}
Err(err) => {
info!("Could not open server-level connection to create {database:?} ({err}); assuming it already exists");
}
}
}
/// `mysql://user:pass@host:port/db?params` → `mysql://user:pass@host:port`
fn strip_database_from_url(url: &str) -> String {
match url.rfind('/') {
// Position 7 is the end of "mysql://" — a URL with no database path.
Some(idx) if idx > "mysql://".len() => url[..idx].to_string(),
_ => url.to_string(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn strips_database_segment() {
assert_eq!(
strip_database_from_url("mysql://u:p@localhost:3306/artcraft_test"),
"mysql://u:p@localhost:3306",
);
}
#[test]
fn leaves_database_free_urls_alone() {
assert_eq!(
strip_database_from_url("mysql://u:p@localhost:3306"),
"mysql://u:p@localhost:3306",
);
}
}
@@ -0,0 +1,184 @@
//! Materializes the full schema into the test database by replaying the
//! checked-in diesel migrations (`_database/sql/migrations/*/up.sql`) in
//! lexical (== chronological) order, then seeding the system roles.
//!
//! Applied migrations are recorded in `mysql_testing_applied_migrations`, so
//! repeat test runs only apply new migrations. If a replay ever wedges the
//! schema, `DROP DATABASE artcraft_test;` and rerun — the harness recreates
//! everything from scratch.
use std::path::{Path, PathBuf};
use log::info;
use sqlx::MySqlPool;
const MIGRATIONS_TABLE_DDL: &str = "
CREATE TABLE IF NOT EXISTS mysql_testing_applied_migrations (
name VARCHAR(255) NOT NULL PRIMARY KEY,
applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
";
/// Bring the test database schema up to date and seed required system rows.
///
/// Single-flighted via a MySQL named lock, so parallel tests (and parallel
/// test processes) can all call this concurrently: one caller applies any
/// pending migrations while the rest wait, then everyone proceeds.
pub async fn ensure_schema(pool: &MySqlPool) {
let mut lock_connection = pool.acquire().await.expect("acquire for schema lock");
let locked: i64 = sqlx::query_scalar("SELECT GET_LOCK('mysql_testing_schema', 120)")
.fetch_one(&mut *lock_connection)
.await
.expect("take schema lock");
assert_eq!(locked, 1, "timed out waiting for the schema lock");
apply_schema(pool).await;
sqlx::query("SELECT RELEASE_LOCK('mysql_testing_schema')")
.execute(&mut *lock_connection)
.await
.expect("release schema lock");
}
async fn apply_schema(pool: &MySqlPool) {
sqlx::query(MIGRATIONS_TABLE_DDL)
.execute(pool)
.await
.expect("create mysql_testing_applied_migrations table");
let applied: Vec<String> =
sqlx::query_scalar("SELECT name FROM mysql_testing_applied_migrations")
.fetch_all(pool)
.await
.expect("read applied migrations");
let migrations_dir = repo_root().join("_database/sql/migrations");
let mut migration_dirs: Vec<PathBuf> = std::fs::read_dir(&migrations_dir)
.unwrap_or_else(|err| panic!("read migrations dir {migrations_dir:?}: {err}"))
.filter_map(|entry| entry.ok())
.map(|entry| entry.path())
.filter(|path| path.is_dir())
.collect();
migration_dirs.sort();
let mut applied_count = 0;
for migration_dir in &migration_dirs {
let name = migration_dir
.file_name()
.and_then(|n| n.to_str())
.expect("migration dir name")
.to_string();
if applied.iter().any(|a| a == &name) {
continue;
}
let up_sql_path = migration_dir.join("up.sql");
if !up_sql_path.exists() {
continue;
}
apply_sql_file(pool, &up_sql_path)
.await
.unwrap_or_else(|err| panic!("migration {name} failed: {err}"));
sqlx::query("INSERT INTO mysql_testing_applied_migrations (name) VALUES (?)")
.bind(&name)
.execute(pool)
.await
.expect("record applied migration");
applied_count += 1;
}
if applied_count > 0 {
info!("Applied {applied_count} migrations to the test database");
}
seed_system_roles(pool).await;
}
/// The `users` table references `user_roles` rows; seed them once.
async fn seed_system_roles(pool: &MySqlPool) {
let role_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM user_roles")
.fetch_one(pool)
.await
.expect("count user_roles");
if role_count > 0 {
return;
}
let seed_path = repo_root().join("_database/sql/seed/sql/system_roles.sql");
apply_sql_file(pool, &seed_path)
.await
.unwrap_or_else(|err| panic!("seeding system roles failed: {err}"));
info!("Seeded system user_roles");
}
/// Execute a file of `;`-separated SQL statements one at a time.
/// (sqlx won't run multi-statement strings; the checked-in migrations contain
/// no stored procedures or DELIMITER blocks, so naive splitting is safe.)
async fn apply_sql_file(pool: &MySqlPool, path: &Path) -> Result<(), sqlx::Error> {
let sql = std::fs::read_to_string(path)
.unwrap_or_else(|err| panic!("read {path:?}: {err}"));
for statement in split_sql_statements(&sql) {
sqlx::query(&statement).execute(pool).await.map_err(|err| {
log::error!("Statement from {path:?} failed:\n{statement}\nerror: {err}");
err
})?;
}
Ok(())
}
/// Split on `;`, dropping `--` line comments and empty statements.
fn split_sql_statements(sql: &str) -> Vec<String> {
let without_line_comments: String = sql
.lines()
.filter(|line| !line.trim_start().starts_with("--"))
.collect::<Vec<_>>()
.join("\n");
without_line_comments
.split(';')
.map(|statement| statement.trim())
.filter(|statement| !statement.is_empty())
.map(|statement| statement.to_string())
.collect()
}
/// The repository root, from this crate's location on disk.
fn repo_root() -> PathBuf {
// crates/schema/database/mysql_testing → repo root is four levels up.
Path::new(env!("CARGO_MANIFEST_DIR"))
.ancestors()
.nth(4)
.expect("repo root")
.to_path_buf()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn repo_root_contains_migrations() {
assert!(repo_root().join("_database/sql/migrations").is_dir());
assert!(repo_root().join("_database/sql/seed/sql/system_roles.sql").is_file());
}
#[test]
fn splits_statements_and_strips_comments() {
let sql = "-- comment\nCREATE TABLE a (id INT);\n\nINSERT INTO a VALUES (1);\n";
let statements = split_sql_statements(sql);
assert_eq!(statements.len(), 2);
assert!(statements[0].starts_with("CREATE TABLE a"));
assert!(statements[1].starts_with("INSERT INTO a"));
}
#[test]
fn empty_and_comment_only_input_yields_no_statements() {
assert!(split_sql_statements("").is_empty());
assert!(split_sql_statements("-- just a comment\n").is_empty());
}
}
@@ -0,0 +1,21 @@
//! Opt-in serialization for database tests that CANNOT be data-isolated.
//!
//! Most database tests should NOT use this: create your own users, wallets,
//! and rows and run in parallel (schema setup already single-flights via a
//! MySQL named lock). Take this lock only when a test must mutate shared or
//! global state (e.g. truncating a table, changing a singleton row):
//!
//! ```ignore
//! let _serial = mysql_testing::serial::acquire_serial_test_lock().await;
//! ```
use std::sync::OnceLock;
use tokio::sync::{Mutex, MutexGuard};
static SERIAL_TEST_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
/// Hold the returned guard for the duration of the test.
pub async fn acquire_serial_test_lock() -> MutexGuard<'static, ()> {
SERIAL_TEST_LOCK.get_or_init(|| Mutex::new(())).lock().await
}
@@ -200,8 +200,19 @@ zip = "0.6.6"
# NB: Only using diesel for migrations support
diesel = { version = "2.2.4", features = ["mysql"] }
[features]
# Database fixture tests (guarded MySQL test database + in-process stub
# providers) RUN BY DEFAULT with `cargo test`. On machines/CI without a local
# MySQL, skip them with:
# SQLX_OFFLINE=true cargo test -p storyteller-web --features skip_database_tests
skip_database_tests = []
[dev-dependencies]
# Testing/Dev Deps: Internal
mysql_testing = { path = "../../../schema/database/mysql_testing" }
# The test dependency graph enables both rustls crypto backends, so tests
# must install one explicitly before anything builds a TLS config.
rustls = { version = "0.23", features = ["aws-lc-rs"] }
test_utils = { path = "../../../testing/test_utils" }
# Testing/Dev Deps: External
@@ -22,6 +22,8 @@
//!
//! Billing math: each file's duration is rounded UP to a whole second, and
//! counted once per reference (the same file referenced twice bills twice).
//! The TOTAL is what gets billed; the router clamps it to the model's
//! billing range (e.g. 4..=30 seconds for Seedance 2.5).
//!
//! Probing NEVER fails a generation: a file whose download or ffprobe fails
//! is billed at the worst-case [`MAX_BILLED_INPUT_SECONDS`] instead (and is
@@ -44,9 +46,12 @@ use tempfile::NamedTempFile;
use tokens::tokens::media_files::MediaFileToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::common_responses::media::media_links_builder::MediaLinksBuilder;
use crate::http_server::endpoints::media_files::helpers::get_media_domain::get_media_domain;
use crate::util::http_download_url_to_tempfile::http_download_url_to_tempfile;
use crate::util::lookup::lookup_media_files_as_cdn_url_list_and_map::apply_media_cdn_override;
/// Downloaded + ffprobed reference videos.
///
@@ -87,6 +92,7 @@ pub async fn fetch_reference_video_sources(
video_tokens: &[MediaFileToken],
http_request: &HttpRequest,
server_environment: ServerEnvironment,
maybe_media_cdn_override_url: Option<&str>,
mysql_connection: &mut PoolConnection<MySql>,
) -> Result<Vec<ReferenceVideoSource>, CommonWebError> {
const CAN_SEE_DELETED: bool = false;
@@ -133,7 +139,12 @@ pub async fn fetch_reference_video_sources(
server_environment,
&bucket_path);
(file.token, media_links.cdn_url.to_string())
let cdn_url = apply_media_cdn_override(
&media_links.cdn_url,
maybe_media_cdn_override_url,
);
(file.token, cdn_url)
})
.collect();
@@ -130,6 +130,7 @@ pub async fn omni_api_image_generate_handler(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
request.image_media_tokens.as_deref().unwrap_or(&[]),
).await?;
@@ -171,6 +171,7 @@ pub async fn omni_api_video_generate_handler(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
@@ -199,6 +200,7 @@ pub async fn omni_api_video_generate_handler(
video_tokens,
&http_request,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&mut mysql_connection,
).await?;
@@ -111,6 +111,7 @@ async fn probe_input_seconds_best_effort(
video_tokens,
http_request,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&mut mysql_connection,
).await {
Ok(sources) => sources,
@@ -133,6 +133,7 @@ pub async fn omni_gen_audio_generate_handler(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
@@ -121,6 +121,7 @@ pub async fn omni_gen_image_generate_handler(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
request.image_media_tokens.as_deref().unwrap_or(&[]),
).await?;
@@ -126,6 +126,7 @@ pub async fn omni_gen_mesh_generate_handler(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
@@ -131,6 +131,7 @@ pub async fn omni_gen_splat_generate_handler(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
@@ -3,3 +3,9 @@ pub mod first_party_minimax_h3;
pub mod helpers;
pub mod pipeline_v2;
pub mod insert_db_job;
// Database fixture tests. Always compiled with `cfg(test)` so IDEs discover
// them and they can't bit-rot, but each test is `ignore`d unless the
// `database_tests` feature is enabled (see tests/mod.rs).
#[cfg(test)]
mod tests;
@@ -154,6 +154,7 @@ pub async fn omni_gen_video_generate_handler(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
@@ -182,6 +183,7 @@ pub async fn omni_gen_video_generate_handler(
video_tokens,
&http_request,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&mut mysql_connection,
).await?;
@@ -0,0 +1,29 @@
//! Database fixture tests for the omni_gen video generate endpoint.
//!
//! These are DATABASE TESTS: they connect to a guarded MySQL test database
//! (see the `mysql_testing` crate — never production, never the local dev
//! database) and drive the real handler with dummy Actix HTTP requests.
//!
//! They RUN BY DEFAULT under `cargo test` and from the IDE. On machines or
//! CI without a local MySQL, skip them with:
//!
//! ```bash
//! SQLX_OFFLINE=true cargo test -p storyteller-web --features skip_database_tests
//! ```
//!
//! Requirements: a local MySQL with an `artcraft_test` database reachable via
//! `ARTCRAFT_TEST_DATABASE_URL` (default
//! `mysql://root:@localhost:3306/artcraft_test`), plus `ffmpeg`/`ffprobe` on
//! PATH (the stub CDN generates real fixture videos for the input-duration
//! billing tests, and the endpoint probes them).
//!
//! These tests run in PARALLEL: every test creates its own users, wallets,
//! and media rows, schema setup is single-flighted via a MySQL named lock,
//! and the stub Kinovi server is one process-wide instance. A test may only
//! opt out of isolation (mutating shared/global rows) if it takes
//! `mysql_testing::serial::acquire_serial_test_lock()` — none currently do.
pub mod support;
mod seedance_2p0;
mod seedance_2p5;
@@ -0,0 +1,5 @@
//! Seedance 2.0 family pricing tests, split by tier.
mod seedance_2p0_fast_tests;
mod seedance_2p0_mini_tests;
mod seedance_2p0_tests;
@@ -0,0 +1,3 @@
//! Seedance 2.5 family pricing tests.
mod seedance_2p5_tests;
@@ -0,0 +1,777 @@
//! Shared harness for database fixture tests: a `ServerState` wired to the
//! guarded test database, a local stub Kinovi server (so NO external calls
//! ever leave the process), and helpers for driving the generate handler
//! with dummy Actix HTTP requests as a fixture user.
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use actix_web::cookie::Cookie;
use actix_web::test::TestRequest;
use actix_web::web::{Data, Json};
use actix_web::HttpRequest;
use chrono::Utc;
use sqlx::MySqlPool;
use actix_artcraft::sessions::anonymous_visitor_tracking::avt_cookie_manager::AvtCookieManager;
use actix_artcraft::sessions::user_sessions::http_user_session_manager::HttpUserSessionManager;
use actix_helpers::middleware::banned_cidr_filter::banned_cidr_set::BannedCidrSet;
use actix_helpers::middleware::banned_ip_filter::ip_ban_list::ip_ban_list::IpBanList;
use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_video_cost_and_generate_request::OmniGenVideoCostAndGenerateRequest;
use artcraft_api_defs::omni_gen::generate_response::omni_gen_video_generate_response::OmniGenVideoGenerateResponse;
use billing_artcraft_component::utils::artcraft_stripe_config::ArtcraftStripeConfig;
use billing_component::stripe::stripe_config::{
FullUrlOrPath, StripeCheckoutConfigs, StripeConfig, StripeCustomerPortalConfigs, StripeSecrets,
};
use cloud_storage::legacy_bucket_client::LegacyBucketClient;
use elasticsearch::http::transport::Transport;
use elasticsearch::Elasticsearch;
use enums::common::generation::common_video_model::CommonVideoModel;
use kinovi_web_client::requests::kinovi_host::{
ENV_KINOVI_CUSTOM_API_HOST, ENV_KINOVI_CUSTOM_CDN_HOST,
};
use memory_caching::arc_ttl_sieve::ArcTtlSieve;
use memory_caching::single_item_ttl_cache::SingleItemTtlCache;
use mysql_queries::mediators::badge_granter::BadgeGranter;
use mysql_queries::mediators::firehose_publisher::FirehosePublisher;
use mysql_testing::fixtures::users::TestUser;
use redis_caching::redis_ttl_cache::RedisTtlCache;
use server_environment::ServerEnvironment;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
use crate::configs::app_startup::redis_rate_limiters::configure_redis_rate_limiters;
use crate::configs::static_api_tokens::StaticApiTokenSet;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::omni_gen::generate::video::omni_gen_video_generate_handler::omni_gen_video_generate_handler;
use crate::http_server::user_lookup::user_session::session_utils::session_checker::SessionChecker;
use crate::http_server::web_utils::scoped_temp_dir_creator::ScopedTempDirCreator;
use crate::http_server::web_utils::web_opaque_cursor_encoder_v2::WebOpaqueCursorEncoderV2;
use crate::http_server::web_utils::web_sort_key_crypto::WebSortKeyCrypto;
use crate::startup::setup_inference_providers::{
BeebleData, FalData, GmiCloudData, GrokApiData, InferenceProviders, KinoviWebData, OpenAiData,
WorldLabsData,
};
use crate::startup::setup_pager::build_pager;
use crate::startup::setup_static_feature_flags::setup_static_feature_flags;
use crate::state::certs::google_sign_in_cert::GoogleSignInCert;
use crate::state::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache;
use crate::state::server_state::{
Dashboards, DataboxDashboards, DurableInMemoryCaches, EnvConfig, EphemeralInMemoryCaches,
InMemoryCaches, ResendData, ServerInfo, ServerState, StripeSettings, TrollBans,
};
use crate::threads::db_health_checker_thread::db_health_check_status::HealthCheckStatus;
use crate::util::troll_user_bans::troll_user_ban_list::TrollUserBanList;
const TEST_COOKIE_DOMAIN: &str = "localhost";
const TEST_COOKIE_SECRET: &str = "test_cookie_secret";
/// Everything a database test needs. Keep it alive for the test's duration.
pub struct TestHarness {
pub pool: MySqlPool,
pub server_state: Arc<ServerState>,
pub stub_kinovi_base_url: String,
}
impl TestHarness {
/// Connect to the guarded test database, start the stub Kinovi server, and
/// build a ServerState pointing at both.
pub async fn create() -> TestHarness {
// The test binary compiles rustls with BOTH crypto backends (via the
// union of dev-dependencies), which makes TLS-config construction panic
// with an ambiguous-provider error unless one is installed explicitly.
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
// One stub per process; the OnceLock init also points the Kinovi env
// override at it BEFORE any test can issue a request, so tests are free
// to run in parallel (each creates isolated users/wallets/media).
let stub_kinovi_base_url = process_stub_kinovi_base_url().clone();
let pool = mysql_testing::pool::create_test_pool().await;
let server_state = Arc::new(build_test_server_state(pool.clone(), stub_kinovi_base_url.clone()));
TestHarness {
pool,
server_state,
stub_kinovi_base_url,
}
}
/// Create a fixture user with a session and a wallet funded with `credits`.
pub async fn create_funded_user(&self, credits: u64) -> TestUser {
let user = mysql_testing::fixtures::users::create_test_user(&self.pool)
.await
.expect("create test user");
mysql_testing::fixtures::wallets::fund_wallet_banked(&self.pool, &user.user_token, credits)
.await
.expect("fund wallet");
user
}
/// The user's current total wallet balance (banked + monthly).
pub async fn wallet_balance(&self, user: &TestUser) -> u64 {
mysql_testing::fixtures::wallets::artcraft_wallet_balance(&self.pool, &user.user_token)
.await
.expect("read wallet balance")
.map(|balance| balance.total())
.unwrap_or(0)
}
/// All ledger entries for the user's wallet, oldest first.
pub async fn ledger_entries(
&self,
user: &TestUser,
) -> Vec<mysql_testing::fixtures::wallets::LedgerEntry> {
let wallet_token =
mysql_testing::fixtures::wallets::fund_wallet_banked(&self.pool, &user.user_token, 0)
.await
.expect("resolve wallet token");
mysql_testing::fixtures::wallets::wallet_ledger_entries(&self.pool, &wallet_token)
.await
.expect("read ledger")
}
/// POST the request through the real generate handler as `user`.
pub async fn post_generate(
&self,
user: &TestUser,
request: OmniGenVideoCostAndGenerateRequest,
) -> Result<OmniGenVideoGenerateResponse, CommonWebError> {
let http_request = self.authed_request(user);
omni_gen_video_generate_handler(
http_request,
Json(request),
Data::new(self.server_state.clone()),
)
.await
.map(Json::into_inner)
}
/// A dummy HTTP request carrying the user's session cookie.
fn authed_request(&self, user: &TestUser) -> HttpRequest {
let cookie = self
.server_state
.session_cookie_manager
.create_cookie(&user.session_token, &user.user_token)
.expect("create session cookie");
// TestRequest wants an owned cookie with a 'static-compatible lifetime.
let cookie = Cookie::new("session", cookie.value().to_string());
TestRequest::post()
.uri("/v1/omni_gen/generate/video")
.cookie(cookie)
.peer_addr("127.0.0.1:9999".parse().expect("peer addr"))
.to_http_request()
}
}
/// Every pricing test funds a fresh user with this many credits.
pub const STARTING_CREDITS: u64 = 100_000;
/// Newtype wrappers so pricing-table test cases read unambiguously:
/// `(Some(FourEightyP), Seconds(5), Batch(1), ExpectedCredits(39))`.
#[derive(Clone, Copy, Debug)]
pub struct Seconds(pub u16);
#[derive(Clone, Copy, Debug)]
pub struct Batch(pub u16);
#[derive(Clone, Copy, Debug)]
pub struct ExpectedCredits(pub u64);
/// How many credits MORE the variant charges than its base model.
#[derive(Clone, Copy, Debug)]
pub struct CreditsDelta(pub u64);
/// A generate request with only the model set; tests fill in the rest.
/// Idempotency tokens must be unique per call.
pub fn base_generate_request(model: CommonVideoModel) -> OmniGenVideoCostAndGenerateRequest {
OmniGenVideoCostAndGenerateRequest {
idempotency_token: Some(uuid_v4_like()),
model: Some(model),
prompt: Some("a corgi runs through a field of sunflowers".to_string()),
negative_prompt: None,
start_frame_image_media_token: None,
end_frame_image_media_token: None,
reference_image_media_tokens: None,
reference_video_media_tokens: None,
reference_audio_media_tokens: None,
reference_character_tokens: None,
resolution: None,
aspect_ratio: None,
bitrate: None,
quality: None,
duration_seconds: None,
video_batch_count: None,
generate_audio: None,
estimate_only: None,
}
}
/// A unique 32-hex-char idempotency token (accepted UUID format).
fn uuid_v4_like() -> String {
use std::sync::atomic::{AtomicU64, Ordering};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let count = COUNTER.fetch_add(1, Ordering::Relaxed);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("clock")
.as_nanos();
format!("{:016x}{:016x}", nanos as u64, count)
}
/// The process-wide stub server's base URL. First use spawns the stub
/// (plain OS threads — it outlives every per-test tokio runtime) and sets
/// BOTH the Kinovi host-override and media-CDN-override env vars exactly
/// once, before returning.
fn process_stub_kinovi_base_url() -> &'static String {
use std::sync::OnceLock;
static STUB: OnceLock<String> = OnceLock::new();
STUB.get_or_init(|| {
let base_url = spawn_stub_server();
std::env::set_var(ENV_KINOVI_CUSTOM_API_HOST, &base_url);
std::env::set_var(ENV_KINOVI_CUSTOM_CDN_HOST, &base_url);
base_url
})
}
/// A stub for BOTH the Kinovi API and the media CDN. Runs on plain OS
/// threads so it is independent of any tokio runtime and serves connections
/// until the test process exits.
///
/// - `GET` paths containing `dur{millis}ms` serve a REAL generated video of
/// that duration (ffmpeg, cached per duration) — the reference-video
/// billing probe downloads and ffprobes these.
/// - `POST …uploads.signedUploadUrl` answers with an upload URL back at this
/// stub; the subsequent `PUT` is accepted with 200.
/// - Any other `POST` answers as a successful workflow.runTask with
/// process-unique order ids.
fn spawn_stub_server() -> String {
use std::io::{Read, Write};
use std::net::TcpListener;
use std::sync::atomic::{AtomicU64, Ordering};
// Order/task ids must be unique: generic_inference_jobs has a UNIQUE key
// on the external third-party id — and the test database PERSISTS across
// runs, so ids must be unique across processes, not just within one.
static ORDER_COUNTER: AtomicU64 = AtomicU64::new(1);
let process_prefix = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("clock")
.as_nanos() as u64;
let listener = TcpListener::bind("127.0.0.1:0").expect("bind stub server");
let port = listener.local_addr().expect("local addr").port();
let base_url = format!("http://127.0.0.1:{port}");
let base_url_for_thread = base_url.clone();
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(mut stream) = stream else { continue };
let base_url = base_url_for_thread.clone();
std::thread::spawn(move || {
let _ = stream.set_read_timeout(Some(Duration::from_millis(500)));
let mut buffer = vec![0u8; 1 << 20];
let mut read_total = 0;
loop {
match stream.read(&mut buffer[read_total..]) {
Ok(n) if n > 0 => {
read_total += n;
if buffer[..read_total].windows(4).any(|w| w == b"\r\n\r\n") && read_total > 4 {
break;
}
}
_ => break,
}
}
let head = String::from_utf8_lossy(&buffer[..read_total]).to_string();
let request_line = head.lines().next().unwrap_or_default().to_string();
let (status, content_type, body): (&str, &str, Vec<u8>) =
if request_line.starts_with("GET") {
match extract_fixture_duration_millis(&request_line) {
Some(millis) => ("200 OK", "video/mp4", fixture_video_bytes(millis)),
None => ("404 Not Found", "text/plain", b"not found".to_vec()),
}
} else if request_line.starts_with("PUT") {
("200 OK", "text/plain", Vec::new())
} else if request_line.contains("uploads.signedUploadUrl") {
let upload_url = format!("{base_url}/upload-target/materials/test/fixture.mp4");
let json = format!(r#"[{{"result":{{"data":{{"json":"{upload_url}"}}}}}}]"#);
("200 OK", "application/json", json.into_bytes())
} else {
let order_number = ORDER_COUNTER.fetch_add(1, Ordering::Relaxed);
let json = format!(
r#"[{{"result":{{"data":{{"json":{{"taskId":"task_test_{process_prefix:016x}_{order_number:04}","orderId":"ord_test_{process_prefix:016x}_{order_number:04}","violationWarning":false}}}}}}}}]"#,
);
("200 OK", "application/json", json.into_bytes())
};
let response = format!(
"HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len(),
);
let _ = stream.write_all(response.as_bytes());
let _ = stream.write_all(&body);
let _ = stream.shutdown(std::net::Shutdown::Both);
});
}
});
base_url
}
/// `GET /media/.../dur2500ms_abc.mp4 HTTP/1.1` → `Some(2500)`.
fn extract_fixture_duration_millis(request_line: &str) -> Option<u64> {
let idx = request_line.find("dur")?;
let rest = &request_line[idx + 3..];
let end = rest.find("ms")?;
rest[..end].parse().ok()
}
/// A real MP4 of the requested duration, generated with ffmpeg once per
/// duration and cached on disk for the life of the machine's temp dir.
fn fixture_video_bytes(duration_millis: u64) -> Vec<u8> {
use std::sync::Mutex;
// Parallel tests race on cache generation; serialize it and land finished
// files with an atomic rename so a reader can never see a partial write.
static GENERATION_LOCK: Mutex<()> = Mutex::new(());
let cache_dir = std::env::temp_dir().join("artcraft_mysql_testing_fixture_videos");
std::fs::create_dir_all(&cache_dir).expect("create fixture video cache dir");
let cache_path = cache_dir.join(format!("fixture_{duration_millis}.mp4"));
if !cache_path.exists() {
let _guard = GENERATION_LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
if !cache_path.exists() {
let scratch_path = cache_dir.join(format!(
"fixture_{duration_millis}.partial.{}.mp4",
std::process::id(),
));
let seconds = duration_millis as f64 / 1_000.0;
let status = std::process::Command::new("ffmpeg")
.args([
"-y",
"-f", "lavfi",
// Rate INSIDE the filter so the frame count (and thus the container
// duration) is exact; output retiming via `-r` pads the duration.
"-i", &format!("color=c=black:s=64x64:r=10:d={seconds}"),
"-pix_fmt", "yuv420p",
])
.arg(&scratch_path)
.output()
.expect("ffmpeg must be installed to generate fixture videos for database tests");
assert!(
status.status.success(),
"ffmpeg failed generating a {duration_millis}ms fixture video: {}",
String::from_utf8_lossy(&status.stderr),
);
std::fs::rename(&scratch_path, &cache_path).expect("land fixture video");
}
}
std::fs::read(&cache_path).expect("read fixture video")
}
/// A ServerState whose only live dependencies are the test MySQL pool and
/// the stub Kinovi server. Everything else is inert dummy configuration:
/// Redis/Elasticsearch/Stripe/bucket clients are constructed lazily and are
/// never called on the video generate path.
fn build_test_server_state(pool: MySqlPool, media_cdn_override_url: String) -> ServerState {
let session_cookie_manager = HttpUserSessionManager::new(TEST_COOKIE_DOMAIN, TEST_COOKIE_SECRET)
.expect("session cookie manager");
let avt_cookie_manager =
AvtCookieManager::new(TEST_COOKIE_DOMAIN, TEST_COOKIE_SECRET).expect("avt cookie manager");
// NB: `new` (not `new_with_cache`) keeps Redis out of session checks.
let session_checker = SessionChecker::new(&session_cookie_manager);
let firehose_publisher = FirehosePublisher { mysql_pool: pool.clone() };
let badge_granter = BadgeGranter {
mysql_pool: pool.clone(),
firehose_publisher: firehose_publisher.clone(),
};
// Lazy client: never connects unless used (it isn't on this path).
let redis_pool = r2d2::Pool::builder()
.build_unchecked(redis::Client::open("redis://127.0.0.1:1/").expect("redis client"));
let redis_ttl_cache = RedisTtlCache::new_with_ttl(redis_pool.clone(), 60);
let server_environment = ServerEnvironment::Development;
let (pager, _pager_worker, paging_flags) = build_pager(server_environment, "test-host");
// Built once per process: these construct HTTP/TLS clients that load the
// platform certificate store, which intermittently fails (macOS keychain
// I/O errors) when dozens of parallel tests do it simultaneously.
struct ExpensiveDummies {
stripe: StripeSettings,
stripe_artcraft: billing_artcraft_component::utils::artcraft_stripe_config::ArtcraftStripeConfigWithClient,
elasticsearch: Elasticsearch,
private_bucket_client: LegacyBucketClient,
public_bucket_client: LegacyBucketClient,
auto_gc_bucket_client: LegacyBucketClient,
redis_rate_limiters: crate::state::server_state::RedisRateLimiters,
}
use std::sync::OnceLock;
static DUMMIES: OnceLock<ExpensiveDummies> = OnceLock::new();
let dummies = DUMMIES.get_or_init(|| {
let dummy_bucket = || {
LegacyBucketClient::create(
"test-access-key",
"test-secret-key",
"auto",
"test-bucket",
"http://127.0.0.1:1",
None,
Some(Duration::from_secs(5)),
)
.expect("bucket client")
};
ExpensiveDummies {
stripe: StripeSettings {
config: StripeConfig {
checkout: StripeCheckoutConfigs {
success_url: FullUrlOrPath::Path("/success".to_string()),
cancel_url: FullUrlOrPath::Path("/cancel".to_string()),
},
portal: StripeCustomerPortalConfigs {
return_url: FullUrlOrPath::Path("/portal".to_string()),
default_portal_config_id: "bpc_test".to_string(),
},
secrets: StripeSecrets {
publishable_key: None,
secret_key: "sk_test_dummy".to_string(),
secret_webhook_signing_key: "whsec_test_dummy".to_string(),
},
},
client: stripe::Client::new("sk_test_dummy"),
stripe_account_id: "acct_test".to_string(),
},
stripe_artcraft: ArtcraftStripeConfig {
stripe_account_id: "acct_test".to_string(),
secret_key: "sk_test_dummy".to_string(),
secret_webhook_signing_key: "whsec_test_dummy".to_string(),
checkout_success_url: "http://localhost/success".to_string(),
checkout_cancel_url: "http://localhost/cancel".to_string(),
portal_return_url: "http://localhost/portal".to_string(),
}
.to_config_with_client(),
elasticsearch: Elasticsearch::new(
Transport::single_node("http://127.0.0.1:1").expect("es transport"),
),
private_bucket_client: dummy_bucket(),
public_bucket_client: dummy_bucket(),
auto_gc_bucket_client: dummy_bucket(),
redis_rate_limiters: configure_redis_rate_limiters().expect("rate limiters"),
}
});
ServerState {
env_config: EnvConfig {
num_workers: 1,
bind_address: "127.0.0.1:0".to_string(),
cookie_domain: TEST_COOKIE_DOMAIN.to_string(),
cookie_secure: false,
cookie_http_only: true,
website_homepage_redirect: "http://localhost/".to_string(),
},
server_info: ServerInfo { build_sha: "test".to_string() },
stripe: dummies.stripe.clone(),
stripe_artcraft: dummies.stripe_artcraft.clone(),
hostname: "test-host".to_string(),
startup_time: Utc::now(),
server_environment,
// Server-side media downloads (reference probing, upload resolution) hit
// the stub instead of the real CDN.
maybe_media_cdn_override_url: Some(media_cdn_override_url),
flags: setup_static_feature_flags(paging_flags).expect("feature flags"),
third_party_url_redirector: ThirdPartyUrlRedirector::new(server_environment),
health_check_status: HealthCheckStatus::new(),
mysql_pool: pool,
elasticsearch: dummies.elasticsearch.clone(),
redis_pool,
redis_ttl_cache,
redis_rate_limiters: dummies.redis_rate_limiters.clone(),
session_cookie_manager,
avt_cookie_manager,
session_checker,
firehose_publisher,
badge_granter,
private_bucket_client: dummies.private_bucket_client.clone(),
public_bucket_client: dummies.public_bucket_client.clone(),
auto_gc_bucket_client: dummies.auto_gc_bucket_client.clone(),
seedance_video_bucket: None,
inference_providers: InferenceProviders {
fal: FalData {
api_key: fal_client::creds::fal_api_key::FalApiKey::new("test".to_string()),
webhook_url: "http://localhost/webhook".to_string(),
},
gmicloud: GmiCloudData {
api_key: gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey::new("test".to_string()),
},
grok_api: GrokApiData {
api_key: grok_api_client::creds::grok_api_key::GrokApiKey::new("test".to_string()),
},
beeble: BeebleData {
api_key: beeble_client::creds::beeble_api_key::BeebleApiKey::new("test".to_string()),
webhook_url: "http://localhost/webhook".to_string(),
},
kinovi_web: KinoviWebData {
cookies_volcengine: "test_cookie_volcengine=1".to_string(),
cookies_byteplus: "test_cookie_byteplus=1".to_string(),
cookies_byteplus_ultra: "test_cookie_byteplus_ultra=1".to_string(),
},
openai: OpenAiData { api_key: "test".to_string() },
worldlabs: WorldLabsData { api_key: "test".to_string() },
},
resend: ResendData { api_key: "test".to_string() },
pager,
audio_uploads_bucket_root: "test-audio-uploads".to_string(),
sort_key_crypto: WebSortKeyCrypto::new("test-sort-key"),
opaque_cursors: WebOpaqueCursorEncoderV2::new("test-sort-key"),
ip_ban_list: IpBanList::new(),
cidr_ban_set: BannedCidrSet::new(),
troll_bans: TrollBans {
user_tokens: TrollUserBanList::new(),
ip_addresses: IpBanList::new(),
},
static_api_token_set: StaticApiTokenSet::from_file("/nonexistent/static_api_tokens.toml"),
internal_api_keys: HashSet::new(),
caches: InMemoryCaches {
durable: DurableInMemoryCaches {
model_token_info: ModelTokenToInfoCache::new(),
},
ephemeral: EphemeralInMemoryCaches {
tts_model_list: SingleItemTtlCache::create_with_duration(Duration::from_secs(60)),
database_tts_category_list: SingleItemTtlCache::create_with_duration(Duration::from_secs(60)),
tts_model_category_assignments: SingleItemTtlCache::create_with_duration(Duration::from_secs(60)),
inference_queue_length: SingleItemTtlCache::create_with_duration(Duration::from_secs(60)),
featured_media_files_sieve: ArcTtlSieve::with_capacity_and_ttl_duration(
25,
Duration::from_secs(60),
)
.expect("sieve"),
},
},
google_sign_in_cert: GoogleSignInCert::new(),
temp_dir_creator: ScopedTempDirCreator::auto_setup(),
dashboards: Dashboards {
databox: DataboxDashboards {
daus_id: None,
daily_generations_id: None,
},
},
}
}
/// Fund a fresh user, run one generation to completion via the stub Kinovi
/// server, and assert the wallet was debited exactly the expected credits
/// (balance delta AND ledger entry).
pub async fn assert_successful_generation_charges(
harness: &TestHarness,
model: CommonVideoModel,
resolution: Option<enums::common::generation::common_resolution::CommonResolution>,
Seconds(duration_seconds): Seconds,
Batch(batch_count): Batch,
ExpectedCredits(expected_credits): ExpectedCredits,
) {
let user = harness.create_funded_user(STARTING_CREDITS).await;
let mut request = base_generate_request(model);
request.resolution = resolution;
request.duration_seconds = Some(duration_seconds);
request.video_batch_count = Some(batch_count);
let response = harness
.post_generate(&user, request)
.await
.unwrap_or_else(|err| {
panic!("{:?} {:?} {}s x{}: generation failed: {:?}", model, resolution, duration_seconds, batch_count, err)
});
assert!(response.success);
let balance = harness.wallet_balance(&user).await;
assert_eq!(
STARTING_CREDITS - balance,
expected_credits,
"{:?} {:?} {}s x{}: wrong wallet debit", model, resolution, duration_seconds, batch_count,
);
let entries = harness.ledger_entries(&user).await;
let debit = entries
.iter()
.find(|entry| entry.credits_delta < 0)
.unwrap_or_else(|| panic!("{:?}: no debit ledger entry found", model));
assert_eq!(
-debit.credits_delta,
expected_credits as i64,
"{:?} {:?} {}s x{}: wrong ledger debit", model, resolution, duration_seconds, batch_count,
);
assert!(!debit.is_refunded, "{:?}: successful generation must not be refunded", model);
}
/// Reference-video request: the charge lands (pinning the with-references
/// price), then the unreachable fixture media makes the provider upload fail
/// and the charge is refunded. Asserts the exact debit amount on the
/// refunded ledger entry and that the balance is made whole.
pub async fn assert_reference_video_charge_then_refund(
harness: &TestHarness,
model: CommonVideoModel,
resolution: Option<enums::common::generation::common_resolution::CommonResolution>,
Seconds(duration_seconds): Seconds,
Batch(batch_count): Batch,
ExpectedCredits(expected_credits): ExpectedCredits,
) {
let user = harness.create_funded_user(STARTING_CREDITS).await;
let video_token = mysql_testing::fixtures::media_files::create_test_video_media_file(
&harness.pool,
&user.user_token,
Some(6_000),
)
.await
.expect("create video media file fixture");
let mut request = base_generate_request(model);
request.resolution = resolution;
request.duration_seconds = Some(duration_seconds);
request.video_batch_count = Some(batch_count);
request.reference_video_media_tokens = Some(vec![video_token]);
// The upload of the (unreachable) reference video fails after billing, so
// the endpoint errors and the charge is refunded.
let result = harness.post_generate(&user, request).await;
assert!(
result.is_err(),
"{:?}: generation with unreachable reference media should fail", model,
);
let entries = harness.ledger_entries(&user).await;
let debit = entries
.iter()
.find(|entry| entry.credits_delta < 0)
.unwrap_or_else(|| panic!("{:?}: no debit ledger entry found", model));
assert_eq!(
-debit.credits_delta,
expected_credits as i64,
"{:?} {:?} {}s + refs: wrong charged amount", model, resolution, duration_seconds,
);
assert!(
debit.is_refunded,
"{:?}: failed generation must refund the charge", model,
);
assert_eq!(
harness.wallet_balance(&user).await,
STARTING_CREDITS,
"{:?}: refund must make the wallet whole", model,
);
}
/// For models that currently have no execution route: the request must fail
/// cleanly BEFORE any billing happens.
pub async fn assert_generation_fails_and_charges_nothing(
harness: &TestHarness,
model: CommonVideoModel,
Seconds(duration_seconds): Seconds,
) {
let user = harness.create_funded_user(STARTING_CREDITS).await;
let mut request = base_generate_request(model);
request.duration_seconds = Some(duration_seconds);
let result = harness.post_generate(&user, request).await;
assert!(result.is_err(), "{:?}: unroutable model must be rejected", model);
assert_eq!(
harness.wallet_balance(&user).await,
STARTING_CREDITS,
"{:?}: rejected generation must not charge", model,
);
}
/// Assert the variant model charges a premium over the base model: runs one
/// generation on EACH model, asserts both exact charges, and that
/// `variant == base + delta`. (A delta of 0 is allowed where ceil-rounding
/// collapses a sub-cent premium.)
pub async fn assert_variant_charges_premium(
harness: &TestHarness,
base_model: CommonVideoModel,
variant_model: CommonVideoModel,
resolution: Option<enums::common::generation::common_resolution::CommonResolution>,
seconds: Seconds,
batch: Batch,
base: ExpectedCredits,
variant: ExpectedCredits,
CreditsDelta(delta): CreditsDelta,
) {
let ExpectedCredits(base_credits) = base;
let ExpectedCredits(variant_credits) = variant;
// The table itself must be internally consistent.
assert_eq!(
base_credits + delta,
variant_credits,
"test table bug: {:?} {} + delta {} != {:?} {}",
base_model, base_credits, delta, variant_model, variant_credits,
);
assert_successful_generation_charges(harness, base_model, resolution, seconds, batch, base).await;
assert_successful_generation_charges(harness, variant_model, resolution, seconds, batch, variant).await;
}
/// Run one generation with REAL probeable reference videos (served by the
/// stub CDN with exact durations) and assert the exact wallet debit. The
/// whole flow succeeds: probe → billing → upload (to the stub) → generate.
pub async fn assert_real_input_videos_charge(
harness: &TestHarness,
model: CommonVideoModel,
resolution: Option<enums::common::generation::common_resolution::CommonResolution>,
Seconds(duration_seconds): Seconds,
input_video_millis: &[u64],
ExpectedCredits(expected_credits): ExpectedCredits,
) {
let user = harness.create_funded_user(STARTING_CREDITS).await;
let mut video_tokens = Vec::new();
for millis in input_video_millis {
let token = mysql_testing::fixtures::media_files::create_probeable_test_video_media_file(
&harness.pool,
&user.user_token,
*millis,
)
.await
.expect("create probeable video media file fixture");
video_tokens.push(token);
}
let mut request = base_generate_request(model);
request.resolution = resolution;
request.duration_seconds = Some(duration_seconds);
request.reference_video_media_tokens = Some(video_tokens);
let response = harness
.post_generate(&user, request)
.await
.unwrap_or_else(|err| {
panic!(
"{:?} {:?} {}s inputs {:?}ms: generation failed: {:?}",
model, resolution, duration_seconds, input_video_millis, err,
)
});
assert!(response.success);
let balance = harness.wallet_balance(&user).await;
assert_eq!(
STARTING_CREDITS - balance,
expected_credits,
"{:?} {:?} {}s inputs {:?}ms: wrong wallet debit",
model, resolution, duration_seconds, input_video_millis,
);
let entries = harness.ledger_entries(&user).await;
let debit = entries
.iter()
.find(|entry| entry.credits_delta < 0)
.unwrap_or_else(|| panic!("{:?}: no debit ledger entry found", model));
assert!(!debit.is_refunded, "{:?}: successful generation must not be refunded", model);
}
@@ -215,6 +215,7 @@ pub async fn setup_dependencies(server_hostname: &str) -> AnyhowResult<SetupResu
hostname: server_hostname.to_string(),
startup_time,
server_environment,
maybe_media_cdn_override_url: None, // NB: This is a test escape hatch
flags: service_feature_flags,
third_party_url_redirector,
health_check_status,
@@ -60,6 +60,12 @@ pub struct ServerState {
/// Knowing if we're in production will allow us to turn off development-only functionalities.
pub server_environment: ServerEnvironment,
/// Optional base URL (scheme + host, no trailing slash) that REPLACES the
/// media CDN host wherever the server itself downloads media (reference
/// probing, provider upload resolution). For tests and local development
/// against a stub server; None in production.
pub maybe_media_cdn_override_url: Option<String>,
/// Feature flags will allow us to restart the service with different conditions embedded in the code.
pub flags: StaticFeatureFlags,
@@ -27,6 +27,7 @@ pub async fn lookup_media_files_as_cdn_url_list_and_map(
http_request: &HttpRequest,
mysql_connection: &mut PoolConnection<MySql>,
server_environment: ServerEnvironment,
maybe_media_cdn_override_url: Option<&str>,
tokens: &[MediaFileToken],
) -> Result<MediaFilesAsCdnUrlListAndMap, CommonWebError> {
const CAN_SEE_DELETED: bool = false;
@@ -92,7 +93,10 @@ pub async fn lookup_media_files_as_cdn_url_list_and_map(
server_environment,
&public_bucket_path);
token_to_url_map.insert(file.token, media_links.cdn_url.to_string());
token_to_url_map.insert(
file.token,
apply_media_cdn_override(&media_links.cdn_url, maybe_media_cdn_override_url),
);
}
// In requested order, with duplicates preserved.
@@ -106,3 +110,36 @@ pub async fn lookup_media_files_as_cdn_url_list_and_map(
token_to_url_map,
})
}
/// Rewrite a CDN URL onto the override base (keeping the path) when an
/// override is configured; otherwise pass the URL through unchanged.
pub fn apply_media_cdn_override(cdn_url: &url::Url, maybe_override_base: Option<&str>) -> String {
match maybe_override_base {
Some(base) => format!("{}{}", base.trim_end_matches('/'), cdn_url.path()),
None => cdn_url.to_string(),
}
}
#[cfg(test)]
mod tests {
use super::apply_media_cdn_override;
#[test]
fn no_override_passes_the_url_through() {
let url = url::Url::parse("https://cdn.example/media/a/b.mp4").unwrap();
assert_eq!(apply_media_cdn_override(&url, None), "https://cdn.example/media/a/b.mp4");
}
#[test]
fn override_replaces_scheme_and_host_and_keeps_the_path() {
let url = url::Url::parse("https://cdn.example/media/a/b.mp4").unwrap();
assert_eq!(
apply_media_cdn_override(&url, Some("http://127.0.0.1:5555")),
"http://127.0.0.1:5555/media/a/b.mp4",
);
assert_eq!(
apply_media_cdn_override(&url, Some("http://127.0.0.1:5555/")),
"http://127.0.0.1:5555/media/a/b.mp4",
);
}
}