fakecloud_dynamodb/
export_import.rs

1//! Startup bulk-importer for AWS-format DynamoDB S3 exports (DYNAMODB_JSON).
2//!
3//! Reads a local copy of an AWS DynamoDB S3 export (manifests + gzipped data
4//! files) plus the table's `describe-table` JSON, and materialises it as a
5//! fresh table in state. The AWS attribute wire format IS our storage format
6//! (`AttributeValue` is a `serde_json::Value`), so items copy through verbatim
7//! with no type decoding.
8
9use std::collections::HashMap;
10use std::io::Read as _;
11use std::path::Path;
12
13use serde_json::Value;
14
15use crate::state::{DynamoTable, ProvisionedThroughput, SharedDynamoDbState};
16
17/// A single DynamoDB item: attribute name -> typed AWS wire value.
18type Item = HashMap<String, Value>;
19
20/// Outcome of a startup export import.
21///
22/// The import is idempotent: in persistent-storage mode the snapshot is loaded
23/// before this import runs, so on a restart with the same import flags the
24/// target table is already present. Rather than erroring (which would refuse to
25/// boot), the import is skipped and the already-loaded table is left untouched.
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub enum ImportOutcome {
28    /// A fresh table was materialised from the export with `items` items.
29    Imported { table: String, items: usize },
30    /// A table of the same name already existed in state; import was skipped and
31    /// the existing data was left untouched.
32    SkippedExisting { table: String },
33}
34
35/// Import an AWS-format DynamoDB export as a new table.
36///
37/// Idempotent: if the target table already exists in state (for example loaded
38/// from a persisted snapshot on restart), the import is skipped and the existing
39/// data is preserved instead of erroring out.
40pub fn import_aws_export(
41    state: &SharedDynamoDbState,
42    account_id: &str,
43    region: &str,
44    export_dir: &Path,      // folder holding manifest-summary.json etc.
45    describe_table: &Value, // parsed `aws dynamodb describe-table` JSON
46) -> Result<ImportOutcome, String> {
47    // Accept either the full `{"Table": {...}}` dump or a bare table object.
48    let shape = describe_table.get("Table").unwrap_or(describe_table);
49    let table_name = shape["TableName"]
50        .as_str()
51        .ok_or("describe-table: TableName missing or not a string")?
52        .to_string();
53
54    // Idempotent restart guard: if the table is already present (e.g. restored
55    // from a persisted snapshot before this import runs), leave it untouched and
56    // skip rather than error and refuse to boot. Checked before reading the
57    // export so a no-op restart does no work.
58    if table_exists(state, account_id, &table_name) {
59        tracing::warn!(
60            table = %table_name,
61            "skipping DynamoDB export import: table already exists in state; existing data left untouched"
62        );
63        return Ok(ImportOutcome::SkippedExisting { table: table_name });
64    }
65
66    let (items, declared_total) = read_export_items(export_dir)?;
67    let item_count = items.len();
68    // Manifest integrity: the summary's declared grand total must match what was
69    // actually read. Absent field -> skip the check (older/partial manifests).
70    if let Some(declared) = declared_total {
71        if declared != item_count {
72            return Err(format!(
73                "table {table_name}: manifest-summary declares itemCount {declared} but {item_count} items were read (truncated or corrupt export)"
74            ));
75        }
76    }
77    let table = build_table(&table_name, region, account_id, shape, items)?;
78
79    let mut guard = state.write();
80    let account = guard.get_or_create(account_id);
81    // Re-check under the write lock to close any check-then-insert gap and to
82    // handle a genuinely conflicting table gracefully (warn + skip, never panic).
83    if account.tables.contains_key(&table_name) {
84        tracing::warn!(
85            table = %table_name,
86            "skipping DynamoDB export import: table already exists in state; existing data left untouched"
87        );
88        return Ok(ImportOutcome::SkippedExisting { table: table_name });
89    }
90    account.tables.insert(table_name.clone(), table);
91
92    Ok(ImportOutcome::Imported {
93        table: table_name,
94        items: item_count,
95    })
96}
97
98/// True if `account_id` already holds a table named `table_name`.
99fn table_exists(state: &SharedDynamoDbState, account_id: &str, table_name: &str) -> bool {
100    state
101        .read()
102        .get(account_id)
103        .is_some_and(|account| account.tables.contains_key(table_name))
104}
105
106/// Walk the manifests (not a prefix listing) and read every item. Returns the
107/// items plus the summary manifest's declared `itemCount` (if present) so the
108/// caller can verify the grand total.
109fn read_export_items(export_dir: &Path) -> Result<(Vec<Item>, Option<usize>), String> {
110    let summary = read_json(&export_dir.join("manifest-summary.json"))?;
111    let format = summary["exportFormat"].as_str().unwrap_or_default();
112    if format != "DYNAMODB_JSON" {
113        return Err(format!(
114            "unsupported exportFormat {format:?} (only DYNAMODB_JSON is supported)"
115        ));
116    }
117
118    // `manifest-files.json` is JSON Lines (one object per line), each naming a
119    // data file via `dataFileS3Key`.
120    let manifest = read_text(&export_dir.join("manifest-files.json"))?;
121    let mut items = Vec::new();
122    for line in nonempty_lines(&manifest) {
123        let entry: Value =
124            serde_json::from_str(line).map_err(|e| format!("parse manifest-files line: {e}"))?;
125        let s3_key = entry["dataFileS3Key"]
126            .as_str()
127            .ok_or("manifest-files: dataFileS3Key missing")?;
128        // The manifest key uses the standard `AWSDynamoDB/{export-id}/data/<file>`
129        // layout; resolve it against the local export as `<export_dir>/data/<basename>`.
130        let basename = s3_key.rsplit('/').next().unwrap_or(s3_key);
131        let file_items = read_data_file(&export_dir.join("data").join(basename))?;
132        // Per-file integrity: when the manifest entry declares an itemCount it
133        // must match the records actually read from that data file.
134        if let Some(declared) = entry["itemCount"].as_u64() {
135            if declared as usize != file_items.len() {
136                return Err(format!(
137                    "data file {basename}: manifest declares itemCount {declared} but {} records were read (truncated or corrupt export)",
138                    file_items.len()
139                ));
140            }
141        }
142        items.extend(file_items);
143    }
144    let declared_total = summary["itemCount"].as_u64().map(|n| n as usize);
145    Ok((items, declared_total))
146}
147
148/// Read one gzipped data file; each line is a `{"Item": {...}}` record.
149fn read_data_file(path: &Path) -> Result<Vec<Item>, String> {
150    let file = std::fs::File::open(path).map_err(|e| format!("open {}: {e}", path.display()))?;
151    let mut contents = String::new();
152    flate2::read::GzDecoder::new(file)
153        .read_to_string(&mut contents)
154        .map_err(|e| format!("gunzip {}: {e}", path.display()))?;
155
156    // No dedup: AWS exports carry unique primary keys, so a single data file
157    // never contains a duplicate. Colliding keys would both be retained.
158    nonempty_lines(&contents)
159        .map(|line| {
160            let record: Value =
161                serde_json::from_str(line).map_err(|e| format!("parse data line: {e}"))?;
162            record["Item"]
163                .as_object()
164                .map(|obj| obj.clone().into_iter().collect())
165                .ok_or_else(|| "data line missing Item object".to_string())
166        })
167        .collect()
168}
169
170/// Build a fresh `DynamoTable` from a `describe-table` shape and its items.
171fn build_table(
172    name: &str,
173    region: &str,
174    account_id: &str,
175    shape: &Value,
176    items: Vec<Item>,
177) -> Result<DynamoTable, String> {
178    let key_schema =
179        crate::parse_key_schema(&shape["KeySchema"]).map_err(|e| format!("KeySchema: {e}"))?;
180    let attribute_definitions = crate::parse_attribute_definitions(&shape["AttributeDefinitions"])
181        .map_err(|e| format!("AttributeDefinitions: {e}"))?;
182
183    // Guard against a wrong/hand-edited describe-table producing an
184    // ACTIVE-but-key-corrupt table. For each declared key attribute (HASH and,
185    // if present, RANGE) every item must (1) carry the attribute and (2) carry
186    // it with the scalar type declared in AttributeDefinitions -- the same
187    // presence + type parity the normal write path enforces
188    // (validate_key_in_item / check_key_type). A wrong-typed key would let a
189    // correctly-typed read fail to find the row, so the data would appear to
190    // vanish. Fail the whole import so no partial table lands.
191    for elem in &key_schema {
192        if elem.key_type != "HASH" && elem.key_type != "RANGE" {
193            continue;
194        }
195        let role = if elem.key_type == "HASH" {
196            "partition"
197        } else {
198            "sort"
199        };
200        let expected_type = attribute_definitions
201            .iter()
202            .find(|d| d.attribute_name == elem.attribute_name)
203            .map(|d| d.attribute_type.as_str());
204        for item in &items {
205            let Some(value) = item.get(&elem.attribute_name) else {
206                return Err(format!(
207                    "table {name}: item missing {role} key attribute {:?} declared in KeySchema: {item:?}",
208                    elem.attribute_name
209                ));
210            };
211            if let Some(expected) = expected_type {
212                let actual = crate::state::attribute_type_and_value(value).map(|(ty, _)| ty);
213                if actual != Some(expected) {
214                    return Err(format!(
215                        "table {name}: {role} key attribute {:?} has type {} but AttributeDefinitions declares {expected}: {item:?}",
216                        elem.attribute_name,
217                        actual.unwrap_or("<none>"),
218                    ));
219                }
220            }
221        }
222    }
223
224    // BillingModeSummary wins over a bare BillingMode; default to PROVISIONED.
225    let billing_mode = shape["BillingModeSummary"]["BillingMode"]
226        .as_str()
227        .or_else(|| shape["BillingMode"].as_str())
228        .unwrap_or("PROVISIONED")
229        .to_string();
230    let provisioned_throughput = if billing_mode == "PAY_PER_REQUEST" {
231        ProvisionedThroughput {
232            read_capacity_units: 0,
233            write_capacity_units: 0,
234        }
235    } else {
236        crate::parse_provisioned_throughput(&shape["ProvisionedThroughput"])
237            .map_err(|e| format!("ProvisionedThroughput: {e}"))?
238    };
239
240    let mut table = DynamoTable {
241        name: name.to_string(),
242        arn: format!("arn:aws:dynamodb:{region}:{account_id}:table/{name}"),
243        table_id: uuid::Uuid::new_v4().to_string().replace('-', ""),
244        key_schema,
245        attribute_definitions,
246        provisioned_throughput,
247        items,
248        gsi: crate::parse_gsi(&shape["GlobalSecondaryIndexes"], &billing_mode),
249        lsi: crate::parse_lsi(&shape["LocalSecondaryIndexes"]),
250        tags: crate::parse_tags(&shape["Tags"]),
251        created_at: chrono::Utc::now(),
252        status: "ACTIVE".to_string(),
253        item_count: 0,
254        size_bytes: 0,
255        billing_mode,
256        ttl_attribute: None,
257        ttl_enabled: false,
258        resource_policy: None,
259        pitr_enabled: false,
260        kinesis_destinations: Vec::new(),
261        contributor_insights_status: "DISABLED".to_string(),
262        contributor_insights_counters: std::collections::BTreeMap::new(),
263        stream_enabled: false,
264        stream_view_type: None,
265        stream_arn: None,
266        stream_records: std::sync::Arc::new(parking_lot::RwLock::new(Vec::new())),
267        sse_type: None,
268        sse_kms_key_arn: None,
269        deletion_protection_enabled: false,
270        on_demand_throughput: None,
271        table_class: "STANDARD".to_string(),
272    };
273    table.recalculate_stats(); // fills item_count / size_bytes
274    Ok(table)
275}
276
277fn read_text(path: &Path) -> Result<String, String> {
278    std::fs::read_to_string(path).map_err(|e| format!("read {}: {e}", path.display()))
279}
280
281fn read_json(path: &Path) -> Result<Value, String> {
282    serde_json::from_str(&read_text(path)?).map_err(|e| format!("parse {}: {e}", path.display()))
283}
284
285fn nonempty_lines(text: &str) -> impl Iterator<Item = &str> {
286    text.lines().filter(|l| !l.trim().is_empty())
287}
288
289#[cfg(test)]
290mod tests {
291    use super::*;
292    use serde_json::json;
293    use std::path::PathBuf;
294
295    #[test]
296    fn imports_aws_export_into_state() {
297        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
298            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
299        ));
300
301        let fixtures = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
302        let describe = read_json(&fixtures.join("describe-table.json")).unwrap();
303        let export_dir = fixtures.join("export");
304
305        let outcome =
306            import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
307                .expect("import should succeed");
308        assert_eq!(
309            outcome,
310            ImportOutcome::Imported {
311                table: "Music".to_string(),
312                items: 3,
313            }
314        );
315
316        // Table lives in state with the right item count.
317        let mut guard = state.write();
318        let table = guard
319            .get_or_create("123456789012")
320            .tables
321            .get("Music")
322            .expect("table in state");
323        assert_eq!(table.items.len(), 3);
324        assert_eq!(table.item_count, 3);
325
326        // Type preservation: the "Foo" item carries every DynamoDB attribute
327        // type and each one survives verbatim. Complex types are asserted for
328        // FULL equality (not spot-checked) so a serializer regression that
329        // dropped a later list element / set member would be caught.
330        let foo = table
331            .items
332            .iter()
333            .find(|i| i.get("Artist") == Some(&json!({"S": "Foo"})))
334            .expect("item with Artist=Foo");
335
336        // Scalars: whole typed value.
337        assert_eq!(foo.get("str"), Some(&json!({"S": "hello"})));
338        assert_eq!(foo.get("Plays"), Some(&json!({"N": "42"})));
339        assert_eq!(foo.get("bin"), Some(&json!({"B": "aGVsbG8="})));
340        assert_eq!(foo.get("flag"), Some(&json!({"BOOL": true})));
341        assert_eq!(foo.get("nothing"), Some(&json!({"NULL": true})));
342
343        // List (3 mixed elements) and map (2 fields): exact whole-value equality.
344        assert_eq!(
345            foo.get("list"),
346            Some(&json!({"L": [{"S": "nested"}, {"N": "7"}, {"BOOL": false}]}))
347        );
348        assert_eq!(
349            foo.get("map"),
350            Some(&json!({"M": {"genre": {"S": "rock"}, "year": {"N": "1994"}}}))
351        );
352
353        // Sets: assert length AND every member (order is preserved verbatim,
354        // but check membership so a dropped-element regression can't hide).
355        let ss = foo.get("strset").unwrap()["SS"].as_array().unwrap();
356        assert_eq!(ss.len(), 2);
357        assert!(ss.contains(&json!("x")));
358        assert!(ss.contains(&json!("y")));
359
360        let ns = foo.get("numset").unwrap()["NS"].as_array().unwrap();
361        assert_eq!(ns.len(), 2);
362        assert!(ns.contains(&json!("1")));
363        assert!(ns.contains(&json!("2")));
364
365        let bs = foo.get("binset").unwrap()["BS"].as_array().unwrap();
366        assert_eq!(bs.len(), 2);
367        assert!(bs.contains(&json!("YQ==")));
368        assert!(bs.contains(&json!("Yg==")));
369    }
370
371    #[test]
372    fn rejects_items_missing_declared_key_attribute() {
373        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
374            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
375        ));
376
377        let export_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/export");
378
379        // KeySchema names a partition key that the exported items don't carry;
380        // the import must fail rather than materialise a key-corrupt table.
381        let describe = json!({
382            "Table": {
383                "TableName": "Music",
384                "KeySchema": [{ "AttributeName": "DoesNotExist", "KeyType": "HASH" }],
385                "AttributeDefinitions": [
386                    { "AttributeName": "DoesNotExist", "AttributeType": "S" }
387                ],
388                "BillingMode": "PAY_PER_REQUEST"
389            }
390        });
391
392        let err = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
393            .expect_err("import should fail when items lack the declared key attribute");
394        assert!(
395            err.contains("DoesNotExist"),
396            "error should name the missing key attribute: {err}"
397        );
398    }
399
400    #[test]
401    fn rejects_key_attribute_type_mismatch() {
402        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
403            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
404        ));
405
406        let export_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/export");
407
408        // The exported items carry Artist as a string (`{"S": ...}`), but this
409        // describe-table declares the partition key as numeric (`N`). The import
410        // must fail with a type mismatch, matching the normal write path's
411        // check_key_type, rather than materialise a table a correctly-typed read
412        // could never query.
413        let describe = json!({
414            "Table": {
415                "TableName": "MusicTypeMismatch",
416                "KeySchema": [
417                    { "AttributeName": "Artist", "KeyType": "HASH" },
418                    { "AttributeName": "SongTitle", "KeyType": "RANGE" }
419                ],
420                "AttributeDefinitions": [
421                    { "AttributeName": "Artist", "AttributeType": "N" },
422                    { "AttributeName": "SongTitle", "AttributeType": "S" }
423                ],
424                "BillingMode": "PAY_PER_REQUEST"
425            }
426        });
427
428        let err = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
429            .expect_err("import should fail when a key attribute has the wrong type");
430        assert!(
431            err.contains("Artist") && err.contains("type S") && err.contains("declares N"),
432            "error should describe the key type mismatch: {err}"
433        );
434        // Nothing partial landed in state.
435        let mut guard = state.write();
436        assert!(!guard
437            .get_or_create("123456789012")
438            .tables
439            .contains_key("MusicTypeMismatch"));
440    }
441
442    #[test]
443    fn import_is_idempotent_when_table_already_exists() {
444        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
445            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
446        ));
447
448        let fixtures = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
449        let describe = read_json(&fixtures.join("describe-table.json")).unwrap();
450        let export_dir = fixtures.join("export");
451
452        // First import materialises the table (3 items).
453        let first = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
454            .expect("first import should succeed");
455        assert_eq!(
456            first,
457            ImportOutcome::Imported {
458                table: "Music".to_string(),
459                items: 3,
460            }
461        );
462
463        // Replace the loaded data with a single sentinel row, standing in for a
464        // table restored from a persisted snapshot that has since diverged.
465        {
466            let mut guard = state.write();
467            let table = guard
468                .get_or_create("123456789012")
469                .tables
470                .get_mut("Music")
471                .expect("table in state");
472            let mut sentinel: Item = HashMap::new();
473            sentinel.insert("Artist".to_string(), json!({ "S": "SENTINEL" }));
474            sentinel.insert("SongTitle".to_string(), json!({ "S": "only" }));
475            table.items = vec![sentinel];
476            table.recalculate_stats();
477        }
478
479        // Re-running with the same flags must skip (not error, not overwrite).
480        let second = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
481            .expect("re-import should be a no-op, not an error");
482        assert_eq!(
483            second,
484            ImportOutcome::SkippedExisting {
485                table: "Music".to_string(),
486            }
487        );
488
489        // Existing data was left untouched: still the single sentinel row, not
490        // re-populated with the 3 export items.
491        let mut guard = state.write();
492        let table = guard
493            .get_or_create("123456789012")
494            .tables
495            .get("Music")
496            .expect("table still in state");
497        assert_eq!(table.items.len(), 1);
498        assert_eq!(
499            table.items[0].get("Artist"),
500            Some(&json!({ "S": "SENTINEL" }))
501        );
502    }
503
504    #[test]
505    fn rejects_manifest_item_count_mismatch() {
506        use flate2::write::GzEncoder;
507        use flate2::Compression;
508        use std::io::Write as _;
509
510        // Build a throwaway export whose manifest-summary over-declares the item
511        // count (99) relative to the single row actually present, simulating a
512        // truncated/corrupt export.
513        let dir = std::env::temp_dir().join(format!("fc-ddb-import-{}", uuid::Uuid::new_v4()));
514        std::fs::create_dir_all(dir.join("data")).unwrap();
515        std::fs::write(
516            dir.join("manifest-summary.json"),
517            r#"{"version":"2020-06-30","exportFormat":"DYNAMODB_JSON","itemCount":99}"#,
518        )
519        .unwrap();
520        // Per-file entry carries no itemCount, so only the summary-level check
521        // can fire here.
522        std::fs::write(
523            dir.join("manifest-files.json"),
524            r#"{"dataFileS3Key":"AWSDynamoDB/01700000000000-abcd/data/0001.json.gz"}"#,
525        )
526        .unwrap();
527        let data_file = std::fs::File::create(dir.join("data/0001.json.gz")).unwrap();
528        let mut enc = GzEncoder::new(data_file, Compression::default());
529        enc.write_all(b"{\"Item\":{\"Artist\":{\"S\":\"A\"},\"SongTitle\":{\"S\":\"B\"}}}\n")
530            .unwrap();
531        enc.finish().unwrap();
532
533        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
534            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
535        ));
536        let describe = json!({
537            "Table": {
538                "TableName": "MusicTruncated",
539                "KeySchema": [
540                    { "AttributeName": "Artist", "KeyType": "HASH" },
541                    { "AttributeName": "SongTitle", "KeyType": "RANGE" }
542                ],
543                "AttributeDefinitions": [
544                    { "AttributeName": "Artist", "AttributeType": "S" },
545                    { "AttributeName": "SongTitle", "AttributeType": "S" }
546                ],
547                "BillingMode": "PAY_PER_REQUEST"
548            }
549        });
550
551        let err = import_aws_export(&state, "123456789012", "us-east-1", &dir, &describe)
552            .expect_err("import should fail when manifest itemCount disagrees with the data");
553        assert!(
554            err.contains("99") && err.contains("1 items"),
555            "error should report the declared vs actual counts: {err}"
556        );
557
558        let _ = std::fs::remove_dir_all(&dir);
559    }
560
561    #[test]
562    fn rejects_per_file_item_count_mismatch() {
563        use flate2::write::GzEncoder;
564        use flate2::Compression;
565        use std::io::Write as _;
566
567        // manifest-summary agrees with the data (1 item), but the per-file
568        // manifest entry over-declares itemCount (5). The per-file integrity
569        // check must catch it.
570        let dir = std::env::temp_dir().join(format!("fc-ddb-import-{}", uuid::Uuid::new_v4()));
571        std::fs::create_dir_all(dir.join("data")).unwrap();
572        std::fs::write(
573            dir.join("manifest-summary.json"),
574            r#"{"version":"2020-06-30","exportFormat":"DYNAMODB_JSON","itemCount":1}"#,
575        )
576        .unwrap();
577        std::fs::write(
578            dir.join("manifest-files.json"),
579            r#"{"itemCount":5,"dataFileS3Key":"AWSDynamoDB/01700000000000-abcd/data/0001.json.gz"}"#,
580        )
581        .unwrap();
582        let data_file = std::fs::File::create(dir.join("data/0001.json.gz")).unwrap();
583        let mut enc = GzEncoder::new(data_file, Compression::default());
584        enc.write_all(b"{\"Item\":{\"Artist\":{\"S\":\"A\"},\"SongTitle\":{\"S\":\"B\"}}}\n")
585            .unwrap();
586        enc.finish().unwrap();
587
588        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
589            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
590        ));
591        let describe = json!({
592            "Table": {
593                "TableName": "MusicPerFile",
594                "KeySchema": [
595                    { "AttributeName": "Artist", "KeyType": "HASH" },
596                    { "AttributeName": "SongTitle", "KeyType": "RANGE" }
597                ],
598                "AttributeDefinitions": [
599                    { "AttributeName": "Artist", "AttributeType": "S" },
600                    { "AttributeName": "SongTitle", "AttributeType": "S" }
601                ],
602                "BillingMode": "PAY_PER_REQUEST"
603            }
604        });
605
606        let err = import_aws_export(&state, "123456789012", "us-east-1", &dir, &describe)
607            .expect_err("import should fail when a data file's itemCount is wrong");
608        assert!(
609            err.contains("0001.json.gz") && err.contains('5'),
610            "error should name the data file and declared count: {err}"
611        );
612
613        let _ = std::fs::remove_dir_all(&dir);
614    }
615}
fakecloud_dynamodb/export_import.rs

fakecloud_dynamodb/
export_import.rs
