You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: integration_checkpoint.rs
    path of file: ./repos/delta-rs/crates/core/tests
    
    the code of the file until where you have to start completion: #![cfg(feature = "integration_test")]

use chrono::Utc;
use deltalake_core::checkpoints::{cleanup_expired_logs_for, create_checkpoint};
use deltalake_core::kernel::{DataType, PrimitiveType};
use deltalake_core::writer::{DeltaWriter, JsonWriter};
use deltalake_core::{errors::DeltaResult, DeltaOps, DeltaTableBuilder, ObjectStore};
use deltalake_test::utils::*;
use object_store::path::Path;
use serde_json::json;
use serial_test::serial;
use std::time::Duration;
use tokio::time::sleep;

#[tokio::test]
#[serial]
async fn cleanup_metadata_fs_test() -> TestResult {
    let storage = Box::new(LocalStorageIntegration::default());
    let context = IntegrationContext::new(storage)?;
    cleanup_metadata_test(&context).await?;
    Ok(())
}

// Last-Modified for S3 could not be altered by user, hence using system pauses which makes
// test to run longer but reliable
async fn cleanup_metadata_test(context: &IntegrationContext) -> TestResult {
    let table_uri = context.root_uri();
    let log
    
async fn test_issue_1420_cleanup_expired_logs_for() -> DeltaResult<()> {
    let _ = std::fs::remove_dir_all("./tests/data/issue_1420");

    let mut table = DeltaOps::try_from_uri("./tests/data/issue_1420")
        .await?
        .create()
        .with_column(
            "id",
            DataType::Primitive(PrimitiveType::Integer),
            false,
            None,
        )
        .await?;

    let mut writer = JsonWriter::for_table(&table)?;
    writer.write(vec![json!({"id": 1})]).await?;
    writer.flush_and_commit(&mut table).await?; // v1

    let ts = Utc::now(); // use this ts for log retention expiry
    sleep(Duration::from_secs(1)).await;

    writer.write(vec![json!({"id": 2})]).await?;
    writer.flush_and_commit(&mut table).await?; // v2
    assert_eq!(table.version(), 2);

    create_checkpoint(&table).await.unwrap(); // v2.checkpoint.parquet

    // Should delete v1 but not v2 or v2.checkpoint.parquet
    cleanup_expired_logs_for(
        table.version(),
        table.log_store().as_ref(),
        ts.timestamp_millis(),
    )
    .await?;

    assert!(
        table
            .object_store()
            .head(&Path::from(format!("_delta_log/{:020}.json", 1)))
            .await
            .is_err(),
        "commit should not exist"
    );

    assert!(
        table
            .object_store()
            .head(&Path::from(format!("_delta_log/{:020}.json", 2)))
            .await
            .is_ok(),
        "commit should exist"
    );

    assert!(
        table
            .object_store()
            .head(&Path::from(format!(
                "_delta_log/{:020}.checkpoint.parquet",
                2
            )))
            .await
            .is_ok(),
        "checkpoint should exist"
    );

    // pretend time advanced but there is no new versions after v2
    // v2 and v2.checkpoint.parquet should still be there
    let ts = Utc::now();
    sleep(Duration::from_secs(1)).await;

    cleanup_expired_logs_for(
        table.version(),
        table.log_store().as_ref(),
        ts.timestamp_millis(),
    )
    .await?;

    assert!(
        table
            .object_store()
            .head(&Path::from(format!("_delta_log/{:020}.json", 2)))
            .await
            .is_ok(),
        "commit should exist"
    );

    assert!(
        table
            .object_store()
            .head(&Path::from(format!(
                "_delta_log/{:020}.checkpoint.parquet",
                2
            )))
            .await
            .is_ok(),
        "checkpoint should exist"
    );

    Ok(())
}