Introduction to MongoDB: Code

"""Shared storage location and starting documents for the Mongita lesson.

Mongita API and original mongoose example:
https://github.com/scottrogowski/mongita
"""
from pathlib import Path
from mongita import MongitaClientDisk

DATA_DIR = Path(__file__).resolve().parent / "data" / "mongita"


def connect(data_dir=DATA_DIR):
    """Return the client and collection; use a supplied folder in tests."""
    Path(data_dir).mkdir(parents=True, exist_ok=True)
    client = MongitaClientDisk(host=str(data_dir))
    collection = client.hello_world_db.mongoose_collection
    return client, collection


def seed(collection, reset=False):
    """Seed an empty collection, or explicitly reset this lesson collection."""
    if reset:
        collection.delete_many({})
    if collection.count_documents({}) == 0:
        # New dictionaries each time: insertion adds an _id to each document.
        collection.insert_many([
            {"name": "Meercat", "does_not_eat": "Snakes"},
            {"name": "Yellow mongoose", "eats": "Termites"},
        ])

$ pip3 install mongita
Collecting mongita
  Downloading mongita-1.2.0.tar.gz (54 kB)
  Installing build dependencies ... done
  Getting requirements to build wheel ... done
  Preparing metadata (pyproject.toml) ... done
Collecting pymongo<5.0,>=3.0 (from mongita)
  Downloading pymongo-4.11.2-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (22 kB)
Collecting sortedcontainers<3.0,>=2.3 (from mongita)
  Downloading sortedcontainers-2.4.0-py2.py3-none-any.whl.metadata (10 kB)
Collecting dnspython<3.0.0,>=1.16.0 (from pymongo<5.0,>=3.0->mongita)
  Downloading dnspython-2.7.0-py3-none-any.whl.metadata (5.8 kB)
Downloading pymongo-4.11.2-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (1.7 MB)
   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 1.7/1.7 MB 42.8 MB/s eta 0:00:00
Downloading sortedcontainers-2.4.0-py2.py3-none-any.whl (29 kB)
Downloading dnspython-2.7.0-py3-none-any.whl (313 kB)
Building wheels for collected packages: mongita
  Building wheel for mongita (pyproject.toml) ... done
  Created wheel for mongita: filename=mongita-1.2.0-py3-none-any.whl size=61313 sha256=d66120640505c536f50163b98e4d350bcd99e0f717b0c02592461d6c472c1e87
  Stored in directory: /home/codespace/.cache/pip/wheels/c0/bf/0d/6a2a625c704f15045f2e0286f8a819f2104eeaefb92cd42a4f
Successfully built mongita
Installing collected packages: sortedcontainers, dnspython, pymongo, mongita
Successfully installed dnspython-2.7.0 mongita-1.2.0 pymongo-4.11.2 sortedcontainers-2.4.0
$ python
Python 3.12.1 (main, Dec 12 2024, 22:30:56) [GCC 9.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from mongita import MongitaClientDisk
>>> client = MongitaClientDisk()
>>> hello_world_db = client.hello_world_db
>>> mongoose_collection = hello_world_db.mongoose_collection
>>> mongoose_collection.insert_many([{'name': 'Meercat', 'does_not_eat': 'Snakes'},
...                                      {'name': 'Yellow mongoose', 'eats': 'Termites'}])
<mongita.results.InsertManyResult object at 0x72f53e4d6240>
>>> mongoose_collection.count_documents({})
2
>>> mongoose_collection.update_one({'name': 'Meercat'}, {'$set': {"weight": 2}})
<mongita.results.UpdateResult object at 0x72f53d84a600>
>>> mongoose_collection.find({'weight': {'$gt': 1}})
<mongita.cursor.Cursor object at 0x72f53e25faa0>
>>> cursor = mongoose_collection.find({'weight': {'$gt': 1}})
>>> cursor
<mongita.cursor.Cursor object at 0x72f53d881670>
>>> list(cursor)
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('67d8c61b5180a31695e90744'), 'weight': 2}]
>>> len(list(cursor))
0
>>> cursor = mongoose_collection.find({'weight': {'$gt': 1}})
>>> mongoose_list = list(cursor)
>>> len(mongoose_list)
1
>>> mongoose_list
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('67d8c61b5180a31695e90744'), 'weight': 2}]
>>> list(mongoose_collection.find({'weight': {'$gt': 1}}))
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('67d8c61b5180a31695e90744'), 'weight': 2}]
>>> list(mongoose_collection.find())
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('67d8c61b5180a31695e90744'), 'weight': 2}, {'name': 'Yellow mongoose', 'eats': 'Termites', '_id': ObjectId('67d8c61b5180a31695e90745')}]
>>> list(mongoose_collection.find_one())
['name', 'does_not_eat', '_id', 'weight']
>>> list(mongoose_collection.find_one({}))
['name', 'does_not_eat', '_id', 'weight']
>>> list(mongoose_collection.find({}))
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('67d8c61b5180a31695e90744'), 'weight': 2}, {'name': 'Yellow mongoose', 'eats': 'Termites', '_id': ObjectId('67d8c61b5180a31695e90745')}]
>>> list(mongoose_collection.find({'weight': {'$gt': 1}}))
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('67d8c61b5180a31695e90744'), 'weight': 2}]
>>> mongoose_collection.delete_one({'name': 'Meercat'})
<mongita.results.DeleteResult object at 0x72f53d8e0140>
>>> list(mongoose_collection.find({'weight': {'$gt': 1}}))
[]
>>> mongoose_collection.insert_many([{'name': 'Meercat', 'does_not_eat': 'Snakes'}])
<mongita.results.InsertManyResult object at 0x72f53d8e01a0>
>>> list(mongoose_collection.find({'weight': {'$gt': 1}}))
[]
>>> list(mongoose_collection.find())
[{'name': 'Yellow mongoose', 'eats': 'Termites', '_id': ObjectId('67d8c61b5180a31695e90745')}, {'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('67d8c9615180a31695e90748')}]
>>>
$ exit

"""Run the mongoose session with visible output.

Run with --reset to replace the lesson collection before the demonstration.
The original interactive transcripts remain in session.txt and mongita-example.txt.
"""
import argparse
from pprint import pprint
from bson.objectid import ObjectId
from lesson import connect, seed


def demonstrate(collection):
    print("Starting documents:", collection.count_documents({}))
    pprint(list(collection.find({})))

    # The filter selects a document; $set adds or changes just this field.
    result = collection.update_one({"name": "Meercat"}, {"$set": {"weight": 2}})
    print("Matched:", result.matched_count, "Modified:", result.modified_count)
    cursor = collection.find({"weight": {"$gt": 1}})
    mongooses = list(cursor)
    print("First pass:", len(mongooses))
    print("Second pass through cursor:", list(cursor))
    print("Saved list still has:", len(mongooses))

    # A single result is a dictionary, and its identifier is an ObjectId.
    yellow = collection.find_one({"eats": "Termites"})
    identifier = str(yellow["_id"])
    print("String identifier:", identifier)
    pprint(collection.find_one({"_id": ObjectId(identifier)}))
    print("Missing single result:", collection.find_one({"weight": {"$gt": 300}}))
    print("Missing multiple results:", list(collection.find({"weight": {"$gt": 300}})))

    old = collection.find_one({"name": "Meercat"})
    result = collection.delete_one({"_id": old["_id"]})
    print("Deleted:", result.deleted_count)
    inserted = collection.insert_one({"name": "Meercat", "does_not_eat": "Snakes"})
    print("New identifier:", inserted.inserted_id != old["_id"])
    print("Weight query after reinsertion:", list(collection.find({"weight": {"$gt": 1}})))
    print("Final documents:")
    pprint(list(collection.find({})))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--reset", action="store_true",
                        help="delete the lesson documents and run from the beginning")
    args = parser.parse_args()
    if not args.reset:
        parser.error("pass --reset to replace the lesson collection and run the demonstration")
    client, collection = connect()
    try:
        seed(collection, reset=True)
        demonstrate(collection)
    finally:
        client.close()


if __name__ == "__main__":
    main()

from flask import Flask, render_template, request, redirect, url_for
import database

# remember to $ pip install flask
# remember to $ pip install mongita

database.setup_database("pets")

app = Flask(__name__)


def error_page(message, status=400):
    # Simple text response page, as requested.
    return message, status, {"Content-Type": "text/plain; charset=utf-8"}


@app.route("/", methods=["GET"])
@app.route("/list", methods=["GET"])
def get_list():
    pets = database.get_pets()
    return render_template("list.html", pets=pets)


@app.route("/create", methods=["GET"])
def get_create():
    owners = database.get_owners()
    return render_template("create.html", owners=owners)


@app.route("/create", methods=["POST"])
def post_create():
    data = dict(request.form)
    try:
        database.create_pet(data)
        return redirect(url_for("get_list"))
    except (ValueError, database.ConstraintError) as e:
        return error_page(f"Error: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error creating pet: {e}", 500)


@app.route("/delete/<id>", methods=["GET"])
def get_delete(id):
    try:
        database.delete_pet(id)
        return redirect(url_for("get_list"))
    except ValueError as e:
        return error_page(f"Error: {e}", 400)
    except database.NotFoundError as e:
        return error_page(f"Error: {e}", 404)
    except Exception as e:
        return error_page(f"Unexpected error deleting pet: {e}", 500)


@app.route("/update/<id>", methods=["GET"])
def get_update(id):
    try:
        data = database.get_pet(id)
        if data is None:
            return error_page("Error: pet not found.", 404)
        owners = database.get_owners()
        return render_template("update.html", data=data, owners=owners)
    except ValueError as e:
        return error_page(f"Error: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error loading pet: {e}", 500)


@app.route("/update/<id>", methods=["POST"])
def post_update(id):
    data = dict(request.form)
    try:
        database.update_pet(id, data)
        return redirect(url_for("get_list"))
    except (ValueError, database.ConstraintError) as e:
        return error_page(f"Error: {e}", 400)
    except database.NotFoundError as e:
        return error_page(f"Error: {e}", 404)
    except Exception as e:
        return error_page(f"Unexpected error updating pet: {e}", 500)


@app.route("/owners", methods=["GET"])
def get_owners_list():
    owners = database.get_owners()
    return render_template("owner_list.html", owners=owners)


@app.route("/owner/create", methods=["GET"])
def get_owner_create():
    return render_template("owner_create.html")


@app.route("/owner/create", methods=["POST"])
def post_owner_create():
    data = dict(request.form)
    try:
        database.create_owner(data)
        return redirect(url_for("get_owners_list"))
    except ValueError as e:
        return error_page(f"Error: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error creating owner: {e}", 500)


@app.route("/owner/delete/<id>", methods=["GET"])
def get_owner_delete(id):
    try:
        database.delete_owner(id)
        return redirect(url_for("get_owners_list"))
    except ValueError as e:
        return error_page(f"Error: {e}", 400)
    except database.NotFoundError as e:
        return error_page(f"Error: {e}", 404)
    except database.ConstraintError as e:
        return error_page(f"Error: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error deleting owner: {e}", 500)


@app.route("/owner/update/<id>", methods=["GET"])
def get_owner_update(id):
    try:
        data = database.get_owner(id)
        if data is None:
            return error_page("Error: owner not found.", 404)
        return render_template("owner_update.html", data=data)
    except ValueError as e:
        return error_page(f"Error: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error loading owner: {e}", 500)


@app.route("/owner/update/<id>", methods=["POST"])
def post_owner_update(id):
    data = dict(request.form)
    try:
        database.update_owner(id, data)
        return redirect(url_for("get_owners_list"))
    except ValueError as e:
        return error_page(f"Error: {e}", 400)
    except database.NotFoundError as e:
        return error_page(f"Error: {e}", 404)
    except Exception as e:
        return error_page(f"Unexpected error updating owner: {e}", 500)


@app.route("/health", methods=["GET"])
def health():
    try:
        database.get_pets()
        database.get_owners()
        return error_page("ok", 200)
    except Exception as e:
        return error_page(f"Error checking health: {e}", 500)

"""Pets data layer backed by local Mongita documents.

The application uses string IDs; stored _id and owner_id values are ObjectIds.
Mongita: https://github.com/scottrogowski/mongita
ObjectId: https://pymongo.readthedocs.io/en/stable/api/bson/objectid.html
"""
from pathlib import Path
from bson.errors import InvalidId

from bson.objectid import ObjectId
from mongita import MongitaClientDisk

try:
    from mongita import MongitaClientMemory
except ImportError:  # pragma: no cover - some mongita installs may omit this helper
    MongitaClientMemory = None

try:
    import pytest
except ImportError:  # pragma: no cover - only needed when running the tests
    pytest = None

DATA_DIR = Path(__file__).resolve().parent / "data" / "mongita"

client = None
db = None
owners_collection = None
pets_collection = None


class NotFoundError(LookupError):
    pass


class ConstraintError(ValueError):
    pass


def initialize(database_name="pets", client_factory=MongitaClientDisk, data_dir=None):
    global client, db, owners_collection, pets_collection

    if client_factory is MongitaClientMemory and MongitaClientMemory is None:
        if pytest is not None:
            pytest.skip("MongitaClientMemory is not available in this environment.")
        raise RuntimeError("MongitaClientMemory is not available in this environment.")

    close_connection()

    if client_factory is MongitaClientDisk:
        folder = Path(data_dir) if data_dir is not None else DATA_DIR
        folder.mkdir(parents=True, exist_ok=True)
        client = client_factory(host=str(folder))
    else:
        client = client_factory()
    db = client[database_name]
    owners_collection = db.owners
    pets_collection = db.pets


def test_initialize_sets_globals():
    initialize("pytest_initialize", client_factory=MongitaClientMemory)
    assert client is not None
    assert db is not None
    assert owners_collection is not None
    assert pets_collection is not None
    close_connection()


def setup_database(database_name="pets", client_factory=MongitaClientDisk, data_dir=None):
    """
    Prepare the Mongo database and ensure the collections exist.

    Mongita creates collections lazily when they are accessed, so touching the
    collection attributes here is enough to initialize a fresh database.
    """

    initialize(database_name, client_factory=client_factory, data_dir=data_dir)
    owners_collection.count_documents({})
    pets_collection.count_documents({})


def test_setup_database_creates_collections():
    setup_database("pytest_setup", client_factory=MongitaClientMemory)
    assert owners_collection is not None
    assert pets_collection is not None
    assert owners_collection.count_documents({}) == 0
    assert pets_collection.count_documents({}) == 0
    close_connection()


def close_connection():
    global client, db, owners_collection, pets_collection

    if client is not None:
        try:
            client.close()
        except Exception:
            pass
    client = None
    db = None
    owners_collection = None
    pets_collection = None


def test_close_connection_resets_globals():
    initialize("pytest_close", client_factory=MongitaClientMemory)
    close_connection()
    assert client is None
    assert db is None
    assert owners_collection is None
    assert pets_collection is None


def _normalize_age(value):
    if value is None or (isinstance(value, str) and not value.strip()):
        return 0
    try:
        age = int(value)
    except (TypeError, ValueError) as exc:
        raise ValueError("Age must be a whole number.") from exc
    if age < 0:
        raise ValueError("Age must be non-negative.")
    return age


def _require_text(value, field_name):
    text = (value or "").strip()
    if text == "":
        raise ValueError(f"{field_name} is required.")
    return text


def _to_object_id(value, field_name="id"):
    # A URL/form supplies text; an _id filter needs the stored ObjectId type.
    try:
        return ObjectId(str(value))
    except (InvalidId, TypeError) as exc:
        raise ValueError(f"{field_name} must be a valid ObjectId string.") from exc


def _require_owner(owner_id):
    object_id = _to_object_id(owner_id, "owner_id")
    owner = owners_collection.find_one({"_id": object_id})
    if owner is None:
        raise ConstraintError("owner_id does not reference an existing owner.")
    return object_id


def _require_existing_pet(id):
    object_id = _to_object_id(id, "pet id")
    pet = pets_collection.find_one({"_id": object_id})
    if pet is None:
        raise NotFoundError("pet not found.")
    return object_id, pet


def _require_existing_owner(id):
    object_id = _to_object_id(id, "owner id")
    owner = owners_collection.find_one({"_id": object_id})
    if owner is None:
        raise NotFoundError("owner not found.")
    return object_id, owner


def _normalize_pet_data(data):
    owner_id = data.get("owner_id")
    if (owner_id or "").strip() == "":
        raise ValueError("owner_id is required.")

    return {
        "name": _require_text(data.get("name"), "name"),
        "type": _require_text(data.get("type"), "type"),
        "age": _normalize_age(data.get("age")),
        "owner_id": _require_owner(owner_id),
    }


def _normalize_owner_data(data):
    return {
        "name": _require_text(data.get("name"), "name"),
        "city": (data.get("city") or "").strip() or None,
        "type_of_home": (data.get("type_of_home") or "").strip() or None,
    }


def pet_to_dict(pet):
    # Keep BSON identifiers inside this layer; the app receives ordinary strings.
    return {
        "id": str(pet["_id"]),
        "name": pet["name"],
        "type": pet["type"],
        "age": pet["age"],
        "owner_id": str(pet["owner_id"]),
    }


def test_pet_to_dict():
    sample = {
        "_id": ObjectId("67d8c61b5180a31695e90744"),
        "name": "meercat",
        "type": "mammal",
        "age": 2,
        "owner_id": ObjectId("67d8c61b5180a31695e90745"),
    }
    converted = pet_to_dict(sample)
    assert converted["id"] == "67d8c61b5180a31695e90744"
    assert converted["owner_id"] == "67d8c61b5180a31695e90745"
    assert converted["name"] == "meercat"
    assert converted["type"] == "mammal"
    assert converted["age"] == 2


def owner_to_dict(owner):
    return {
        "id": str(owner["_id"]),
        "name": owner["name"],
        "city": owner.get("city"),
        "type_of_home": owner.get("type_of_home"),
    }


def test_owner_to_dict():
    sample = {
        "_id": ObjectId("67d8c61b5180a31695e90746"),
        "name": "greg",
        "city": "Portland",
        "type_of_home": "condo",
    }
    converted = owner_to_dict(sample)
    assert converted["id"] == "67d8c61b5180a31695e90746"
    assert converted["name"] == "greg"
    assert converted["city"] == "Portland"
    assert converted["type_of_home"] == "condo"


def _seed_test_database(database_name="pytest_seed"):
    setup_database(database_name, client_factory=MongitaClientMemory)

    owners_collection.delete_many({})
    pets_collection.delete_many({})

    owners = [
        {"name": "greg", "city": "Portland", "type_of_home": "condo"},
        {"name": "david", "city": "Seattle", "type_of_home": "farm"},
    ]
    owner_ids = {}
    for owner in owners:
        owner_id = create_owner(owner)
        owner_ids[owner["name"]] = owner_id

    pets = [
        {"name": "dorothy", "type": "dog", "age": 9, "owner_id": owner_ids["greg"]},
        {"name": "suzy", "type": "mouse", "age": 9, "owner_id": owner_ids["greg"]},
        {"name": "casey", "type": "dog", "age": 9, "owner_id": owner_ids["greg"]},
        {"name": "heidi", "type": "cat", "age": 15, "owner_id": owner_ids["david"]},
    ]
    for pet in pets:
        create_pet(pet)

    return owner_ids


def get_pets():
    return [pet_to_dict(pet) for pet in pets_collection.find()]


def test_get_pets():
    owner_ids = _seed_test_database()
    pets = get_pets()
    assert type(pets) is list
    assert len(pets) >= 1
    assert type(pets[0]) is dict
    for key in ["id", "name", "type", "age", "owner_id"]:
        assert key in pets[0]
    assert type(pets[0]["id"]) is str
    assert type(pets[0]["owner_id"]) is str
    assert pets[0]["owner_id"] in owner_ids.values()


def get_pet(id):
    object_id = _to_object_id(id, "pet id")
    pet = pets_collection.find_one({"_id": object_id})
    if pet is None:
        return None
    return pet_to_dict(pet)


def test_get_pet():
    owner_ids = _seed_test_database()
    pet = get_pets()[0]
    fetched = get_pet(pet["id"])
    assert fetched is not None
    assert fetched["id"] == pet["id"]
    assert fetched["owner_id"] in owner_ids.values()


def test_get_pet_missing_returns_none():
    _seed_test_database()
    assert get_pet("67d8c61b5180a31695e907ff") is None


def create_pet(data):
    pet = _normalize_pet_data(data)
    result = pets_collection.insert_one(pet)
    # insert_one returns an ObjectId; expose its 24-character text representation.
    return str(result.inserted_id)


def test_create_pet_and_get_pet():
    owner_ids = _seed_test_database()
    new_id = create_pet(
        {
            "name": "walter",
            "age": "2",
            "type": "mouse",
            "owner_id": owner_ids["greg"],
        }
    )
    assert type(new_id) is str
    pet = get_pet(new_id)
    assert pet is not None
    assert pet["id"] == new_id
    assert pet["name"] == "walter"
    assert pet["age"] == 2
    assert pet["type"] == "mouse"
    assert pet["owner_id"] == owner_ids["greg"]


def test_create_pet_requires_name():
    owner_ids = _seed_test_database()
    with pytest.raises(ValueError, match="name is required"):
        create_pet(
            {
                "name": "",
                "age": 1,
                "type": "cat",
                "owner_id": owner_ids["greg"],
            }
        )


def test_create_pet_requires_type():
    owner_ids = _seed_test_database()
    with pytest.raises(ValueError, match="type is required"):
        create_pet(
            {
                "name": "no-type",
                "age": 1,
                "type": "",
                "owner_id": owner_ids["greg"],
            }
        )


def test_create_pet_requires_owner():
    _seed_test_database()
    with pytest.raises(ValueError, match="owner_id is required"):
        create_pet({"name": "ghost", "age": 1, "type": "cat", "owner_id": ""})


def test_create_pet_rejects_unknown_owner():
    _seed_test_database()
    with pytest.raises(ConstraintError, match="owner_id"):
        create_pet(
            {
                "name": "ghost",
                "age": 1,
                "type": "cat",
                "owner_id": "000000000000000000000000",
            }
        )


def test_invalid_pet_id_rejected():
    _seed_test_database()
    with pytest.raises(ValueError, match="ObjectId"):
        get_pet("not-an-object-id")


def update_pet(id, data):
    object_id, _ = _require_existing_pet(id)
    pet = _normalize_pet_data(data)
    pets_collection.update_one({"_id": object_id}, {"$set": pet})


def test_update_pet():
    owner_ids = _seed_test_database()
    pet = get_pets()[0]
    update_pet(
        pet["id"],
        {"name": "updated", "age": "8", "type": "dog", "owner_id": owner_ids["david"]},
    )
    updated = get_pet(pet["id"])
    assert updated is not None
    assert updated["name"] == "updated"
    assert updated["age"] == 8
    assert updated["type"] == "dog"
    assert updated["owner_id"] == owner_ids["david"]


def test_update_pet_rejects_missing_pet():
    owner_ids = _seed_test_database()
    with pytest.raises(NotFoundError, match="pet not found"):
        update_pet(
            "67d8c61b5180a31695e907ff",
            {"name": "updated", "age": 8, "type": "dog", "owner_id": owner_ids["greg"]},
        )


def delete_pet(id):
    object_id, _ = _require_existing_pet(id)
    pets_collection.delete_one({"_id": object_id})


def test_delete_pet():
    owner_ids = _seed_test_database()
    new_id = create_pet(
        {"name": "delete_me", "age": 3, "type": "fish", "owner_id": owner_ids["greg"]}
    )
    delete_pet(new_id)
    assert get_pet(new_id) is None


def test_delete_missing_pet_raises_not_found():
    _seed_test_database()
    with pytest.raises(NotFoundError, match="pet not found"):
        delete_pet("000000000000000000000000")


def get_owners():
    return [owner_to_dict(owner) for owner in owners_collection.find()]


def test_get_owners():
    _seed_test_database()
    owners = get_owners()
    assert type(owners) is list
    assert len(owners) == 2
    assert type(owners[0]) is dict
    for key in ["id", "name", "city", "type_of_home"]:
        assert key in owners[0]


def get_owner(id):
    object_id = _to_object_id(id, "owner id")
    owner = owners_collection.find_one({"_id": object_id})
    if owner is None:
        return None
    return owner_to_dict(owner)


def test_get_owner():
    owner_ids = _seed_test_database()
    owner = get_owner(owner_ids["greg"])
    assert owner is not None
    assert owner["id"] == owner_ids["greg"]
    assert owner["name"] == "greg"
    assert owner["city"] == "Portland"


def test_get_owner_missing_returns_none():
    _seed_test_database()
    assert get_owner("67d8c61b5180a31695e907ff") is None


def create_owner(data):
    owner = _normalize_owner_data(data)
    result = owners_collection.insert_one(owner)
    # insert_one returns an ObjectId; expose its 24-character text representation.
    return str(result.inserted_id)


def test_create_owner_and_get_owner():
    _seed_test_database()
    new_id = create_owner({"name": "solo", "city": "Akron", "type_of_home": "house"})
    assert type(new_id) is str
    owner = get_owner(new_id)
    assert owner is not None
    assert owner["id"] == new_id
    assert owner["name"] == "solo"
    assert owner["city"] == "Akron"
    assert owner["type_of_home"] == "house"


def test_create_owner_requires_name():
    _seed_test_database()
    with pytest.raises(ValueError, match="name is required"):
        create_owner({"name": "", "city": "Akron", "type_of_home": "house"})


def update_owner(id, data):
    object_id, _ = _require_existing_owner(id)
    owner = _normalize_owner_data(data)
    owners_collection.update_one({"_id": object_id}, {"$set": owner})


def test_update_owner():
    owner_ids = _seed_test_database()
    update_owner(
        owner_ids["greg"],
        {"name": "gregory", "city": "Salem", "type_of_home": "cabin"},
    )
    updated = get_owner(owner_ids["greg"])
    assert updated is not None
    assert updated["name"] == "gregory"
    assert updated["city"] == "Salem"
    assert updated["type_of_home"] == "cabin"


def test_update_owner_rejects_missing_owner():
    _seed_test_database()
    with pytest.raises(NotFoundError, match="owner not found"):
        update_owner(
            "67d8c61b5180a31695e907ff",
            {"name": "missing", "city": "Nowhere", "type_of_home": "house"},
        )


def delete_owner(id):
    object_id, _ = _require_existing_owner(id)
    pet = pets_collection.find_one({"owner_id": object_id})
    if pet is not None:
        raise ConstraintError(
            "Cannot delete this owner because they have pets. Please delete their pets first."
        )

    owners_collection.delete_one({"_id": object_id})


def test_delete_owner_restricted():
    owner_ids = _seed_test_database()
    with pytest.raises(ConstraintError, match="have pets"):
        delete_owner(owner_ids["greg"])


def test_delete_owner_then_pet_succeeds():
    _seed_test_database()
    owner_id = create_owner({"name": "solo", "city": "Akron", "type_of_home": "house"})
    pet_id = create_pet(
        {"name": "onepet", "age": 3, "type": "cat", "owner_id": owner_id}
    )

    with pytest.raises(ConstraintError):
        delete_owner(owner_id)

    delete_pet(pet_id)
    delete_owner(owner_id)
    assert get_owner(owner_id) is None


def test_delete_missing_owner_raises_not_found():
    _seed_test_database()
    with pytest.raises(NotFoundError, match="owner not found"):
        delete_owner("000000000000000000000000")


if __name__ == "__main__":
    owner_ids = _seed_test_database("manual_seed")
    assert len(get_pets()) == 4
    assert len(get_owners()) == 2
    assert get_owner(owner_ids["greg"]) is not None
    close_connection()
    print("done.")

flask==3.1.3
mongita==1.2.0
pymongo==4.16.0

<html>

<head></head>

<body>
    This is the create template.
    <form action="/create" method="post">
        <hr />
        <p>Name:<input name="name" /></p>
        <p>Age:<input name="age" /></p>
        <p>Type:<input name="type" /></p>
        <p>Owner:
            <select name="owner_id">
                <option value="">-- Select an Owner --</option>
                {% for owner in owners %}
                <option value="{{ owner['id'] }}">{{ owner['name'] }}</option>
                {% endfor %}
            </select>
        </p>
        <hr />
        <button type="submit">Create</button>
        <a href="/list">Cancel</a>
        <hr />
    </form>
</body>

</html>

<html>
<h3>Pets:</h3>
<table>
    <tr>
        <th>ID</th>
        <th>Name</th>
        <th>Type</th>
        <th>Age</th>
        <th>Owner ID</th>
    </tr>
    {% for pet in pets %}
    <tr>
        <td>{{ pet['id'] }}</td>
        <td>{{ pet['name'] }}</td>
        <td>{{ pet['type'] }}</td>
        <td>{{ pet['age'] }}</td>
        <td>{{ pet['owner_id'] }}</td>
        <td><a href="/delete/{{pet['id']}}">Delete</a></td>
        <td><a href="/update/{{pet['id']}}">Update</a></td>
    </tr>
    {% endfor %}
</table>
<hr />
<a href="/create">Create New Pet</a>
<hr />
<a href="/owners">Manage Owners</a>

</html>

<html>

<head></head>

<body>
    This is the owner create template.
    <form action="/owner/create" method="post">
        <hr />
        <p>Name:<input name="name" /></p>
        <p>City:<input name="city" /></p>
        <p>Type of Home:<input name="type_of_home" /></p>
        <hr />
        <button type="submit">Create</button>
        <a href="/owners">Cancel</a>
        <hr />
    </form>
</body>

</html>

<html>
<h3>Owners:</h3>
<table>
    <tr>
        <th>ID</th>
        <th>Name</th>
        <th>City</th>
        <th>Type of Home</th>
    </tr>
    {% for owner in owners %}
    <tr>
        <td>{{ owner['id'] }}</td>
        <td>{{ owner['name'] }}</td>
        <td>{{ owner['city'] }}</td>
        <td>{{ owner['type_of_home'] }}</td>
        <td><a href="/owner/delete/{{owner['id']}}">Delete</a></td>
        <td><a href="/owner/update/{{owner['id']}}">Update</a></td>
    </tr>
    {% endfor %}
</table>
<hr />
<a href="/owner/create">Create New Owner</a>
<hr />
<a href="/list">Back to Pets</a>

</html>

<html>

<head></head>

<body>
    This is the owner update template.
    <form action="/owner/update/{{data['id']}}" method="post">
        <hr />
        <p>Name:<input name="name" value="{{data['name']}}" /></p>
        <p>City:<input name="city" value="{{data['city'] or ''}}" /></p>
        <p>Type of Home:<input name="type_of_home" value="{{data['type_of_home'] or ''}}" /></p>
        <hr />
        <button type="submit">Update</button>
        <a href="/owners">Cancel</a>
        <hr />
    </form>
</body>

</html>

<html>

<head></head>

<body>
    This is the update template.
    <form action="/update/{{data['id']}}" method="post">
        <hr />
        <p>Name:<input name="name" value="{{data['name']}}" /></p>
        <p>Age:<input name="age" value="{{data['age']}}" /></p>
        <p>Type:<input name="type" value="{{data['type']}}" /></p>
        <p>Owner:
            <select name="owner_id">
                {% for owner in owners %}
                <option value="{{ owner['id'] }}" {% if owner['id'] == data['owner_id'] %}selected{% endif %}>
                    {{ owner['name'] }}
                </option>
                {% endfor %}
            </select>
        </p>
        <hr />
        <button type="submit">Update</button>
        <a href="/list">Cancel</a>
        <hr />
    </form>
</body>

</html>

import re

import pytest

import app as webapp
import database


@pytest.fixture
def client():
    database.setup_database("pytest_ci", client_factory=database.MongitaClientMemory)
    database.owners_collection.delete_many({})
    database.pets_collection.delete_many({})

    with webapp.app.test_client() as test_client:
        yield test_client

    database.close_connection()


def create_owner(client, name="greg", city="Portland", type_of_home="condo"):
    response = client.post(
        "/owner/create",
        data={"name": name, "city": city, "type_of_home": type_of_home},
        follow_redirects=False,
    )
    assert response.status_code == 302
    assert response.headers["Location"].endswith("/owners")


def get_owner_ids():
    owners = database.get_owners()
    return {owner["name"]: owner["id"] for owner in owners}


def extract_first_pet_id(body):
    match = re.search(r"/update/([0-9a-f]{24})", body)
    assert match is not None
    return match.group(1)


def test_health_route(client):
    response = client.get("/health")
    assert response.status_code == 200
    assert response.get_data(as_text=True) == "ok"


def test_owner_crud_flow(client):
    create_owner(client, name="greg", city="Portland", type_of_home="condo")

    response = client.get("/owners")
    body = response.get_data(as_text=True)
    assert response.status_code == 200
    assert "greg" in body
    assert "Portland" in body

    owner_id = get_owner_ids()["greg"]

    response = client.get(f"/owner/update/{owner_id}")
    assert response.status_code == 200
    assert "greg" in response.get_data(as_text=True)

    response = client.post(
        f"/owner/update/{owner_id}",
        data={"name": "gregory", "city": "Salem", "type_of_home": "cabin"},
        follow_redirects=False,
    )
    assert response.status_code == 302
    assert response.headers["Location"].endswith("/owners")

    updated_owner = database.get_owner(owner_id)
    assert updated_owner is not None
    assert updated_owner["name"] == "gregory"
    assert updated_owner["city"] == "Salem"
    assert updated_owner["type_of_home"] == "cabin"


def test_pet_crud_flow(client):
    create_owner(client, name="greg", city="Portland", type_of_home="condo")
    create_owner(client, name="david", city="Seattle", type_of_home="farm")
    owner_ids = get_owner_ids()

    response = client.post(
        "/create",
        data={
            "name": "walter",
            "age": "2",
            "type": "mouse",
            "owner_id": owner_ids["greg"],
        },
        follow_redirects=False,
    )
    assert response.status_code == 302
    assert response.headers["Location"].endswith("/list")

    response = client.get("/list")
    body = response.get_data(as_text=True)
    assert response.status_code == 200
    assert "walter" in body
    assert owner_ids["greg"] in body

    pet_id = extract_first_pet_id(body)

    response = client.get(f"/update/{pet_id}")
    assert response.status_code == 200
    assert "walter" in response.get_data(as_text=True)

    response = client.post(
        f"/update/{pet_id}",
        data={
            "name": "updated",
            "age": "8",
            "type": "dog",
            "owner_id": owner_ids["david"],
        },
        follow_redirects=False,
    )
    assert response.status_code == 302
    assert response.headers["Location"].endswith("/list")

    updated_pet = database.get_pet(pet_id)
    assert updated_pet is not None
    assert updated_pet["name"] == "updated"
    assert updated_pet["age"] == 8
    assert updated_pet["type"] == "dog"
    assert updated_pet["owner_id"] == owner_ids["david"]

    response = client.get(f"/delete/{pet_id}", follow_redirects=False)
    assert response.status_code == 302
    assert response.headers["Location"].endswith("/list")
    assert database.get_pet(pet_id) is None


def test_pet_create_rejects_unknown_owner(client):
    response = client.post(
        "/create",
        data={
            "name": "ghost",
            "age": "1",
            "type": "cat",
            "owner_id": "000000000000000000000000",
        },
    )
    assert response.status_code == 400
    assert "owner_id" in response.get_data(as_text=True)


def test_owner_delete_is_restricted_by_pets(client):
    create_owner(client, name="greg", city="Portland", type_of_home="condo")
    owner_id = get_owner_ids()["greg"]

    client.post(
        "/create",
        data={
            "name": "dorothy",
            "age": "9",
            "type": "dog",
            "owner_id": owner_id,
        },
    )

    response = client.get(f"/owner/delete/{owner_id}")
    assert response.status_code == 400
    assert "have pets" in response.get_data(as_text=True)

    pet = database.get_pets()[0]
    client.get(f"/delete/{pet['id']}")

    response = client.get(f"/owner/delete/{owner_id}", follow_redirects=False)
    assert response.status_code == 302
    assert response.headers["Location"].endswith("/owners")
    assert database.get_owner(owner_id) is None


def test_create_and_update_owner_validation(client):
    response = client.post("/owner/create", data={"name": "", "city": "x", "type_of_home": "y"})
    assert response.status_code == 400
    assert "name is required" in response.get_data(as_text=True)

    create_owner(client, name="greg", city="Portland", type_of_home="condo")
    owner_id = get_owner_ids()["greg"]

    response = client.post(
        f"/owner/update/{owner_id}",
        data={"name": "", "city": "x", "type_of_home": "y"},
    )
    assert response.status_code == 400
    assert "name is required" in response.get_data(as_text=True)


def test_disk_identifiers_and_reopen(tmp_path):
    from bson.objectid import ObjectId

    database.setup_database("pets", data_dir=tmp_path / "mongita")
    try:
        owner_id = database.create_owner({"name": "Alex"})
        pet_id = database.create_pet({
            "name": "Casey", "type": "dog", "age": "3", "owner_id": owner_id,
        })
        assert isinstance(pet_id, str) and len(pet_id) == 24
        stored = database.pets_collection.find_one({"_id": ObjectId(pet_id)})
        assert stored["owner_id"] == ObjectId(owner_id)
        assert isinstance(stored["_id"], ObjectId)
        database.close_connection()
        database.setup_database("pets", data_dir=tmp_path / "mongita")
        assert database.get_pet(pet_id)["owner_id"] == owner_id
        database.update_pet(pet_id, {
            "name": "Casey", "type": "dog", "age": "4", "owner_id": owner_id,
        })
        assert database.get_pet(pet_id)["age"] == 4
        database.delete_pet(pet_id)
        assert database.get_pet(pet_id) is None
    finally:
        database.close_connection()


def test_id_routes_reject_bad_and_missing_ids(client):
    assert client.get("/update/not-an-id").status_code == 400
    assert client.get("/update/000000000000000000000000").status_code == 404


@pytest.mark.parametrize("age", ["-5", "abc"])
def test_invalid_pet_age_create_and_update(client, age):
    create_owner(client)
    owner_id = get_owner_ids()["greg"]
    data = {"name": "Casey", "type": "dog", "age": age, "owner_id": owner_id}
    response = client.post("/create", data=data)
    assert response.status_code == 400
    assert database.get_pets() == []
    pet_id = database.create_pet(dict(data, age="3"))
    response = client.post(f"/update/{pet_id}", data=data)
    assert response.status_code == 400
    assert database.get_pet(pet_id)["age"] == 3


@pytest.mark.parametrize("age", [None, "", "  ", "0", "3"])
def test_valid_pet_age_defaults_and_conversion(client, age):
    create_owner(client)
    owner_id = get_owner_ids()["greg"]
    pet_id = database.create_pet({"name": "Casey", "type": "dog", "age": age, "owner_id": owner_id})
    assert database.get_pet(pet_id)["age"] == (3 if age == "3" else 0)

mongita==1.2.0
pymongo==4.16.0

$ pip install mongita
Collecting mongita
  Downloading mongita-1.2.0.tar.gz (54 kB)
  Installing build dependencies ... done
  Getting requirements to build wheel ... done
  Preparing metadata (pyproject.toml) ... done
Collecting pymongo<5.0,>=3.0 (from mongita)
  Downloading pymongo-4.16.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl.metadata (10.0 kB)
Collecting sortedcontainers<3.0,>=2.3 (from mongita)
  Downloading sortedcontainers-2.4.0-py2.py3-none-any.whl.metadata (10 kB)
Collecting dnspython<3.0.0,>=2.6.1 (from pymongo<5.0,>=3.0->mongita)
  Downloading dnspython-2.8.0-py3-none-any.whl.metadata (5.7 kB)
Downloading pymongo-4.16.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl (1.7 MB)
   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 1.7/1.7 MB 30.6 MB/s  0:00:00
Downloading dnspython-2.8.0-py3-none-any.whl (331 kB)
Downloading sortedcontainers-2.4.0-py2.py3-none-any.whl (29 kB)
Building wheels for collected packages: mongita
  Building wheel for mongita (pyproject.toml) ... done
  Created wheel for mongita: filename=mongita-1.2.0-py3-none-any.whl size=61345 sha256=e7079e880780f1049860726fd100cabb4f0dc4a41c67dcc58038281b81e6521c
  Stored in directory: /home/codespace/.cache/pip/wheels/c0/bf/0d/6a2a625c704f15045f2e0286f8a819f2104eeaefb92cd42a4f
Successfully built mongita
Installing collected packages: sortedcontainers, dnspython, pymongo, mongita
Successfully installed dnspython-2.8.0 mongita-1.2.0 pymongo-4.16.0 sortedcontainers-2.4.0
$ python
Python 3.12.1 (main, Nov 27 2025, 10:47:52) [GCC 13.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from mongita import MongitaClientDisk
>>> client = MongitaClientDisk()
>>> db = client.helloworld_db
>>> mongoose_collection = helloworld_db.mongoose_collection
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
NameError: name 'helloworld_db' is not defined
>>> hello_world_db = client.helloworld_db
>>> mongoose_collection = helloworld_db.mongoose_collection
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
NameError: name 'helloworld_db' is not defined. Did you mean: 'hello_world_db'?
>>> mongoose_collection = hello_world_db.mongoose_collection
>>> mongoose_collection.insert_many([
... {'name': 'Meercat', 'does_not_eat': 'Snakes'},
... {'name': 'Yellow mongoose', 'eats': 'Termites'}
... ])
<mongita.results.InsertManyResult object at 0x7e62c334e450>
>>> mongoose_collection.count_documents({})
2
>>> mongoose_collection.update_one({'name': 'Meercat'}, {'$set': {"weight": 2}})
<mongita.results.UpdateResult object at 0x7e62c3130d70>
>>> mongoose_collection.find({'weight': {'$gt': 1}})
<mongita.cursor.Cursor object at 0x7e62c33dbd10>
>>> cursor = mongoose_collection.find({'weight': {'$gt': 1}})
>>> cursor
<mongita.cursor.Cursor object at 0x7e62c3e718e0>
>>> list(cursor)
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c4617c762b2683dbe52b19'), 'weight': 2}]
>>> for item in cursor:
...   print(item)
...
>>> cursor = mongoose_collection.find({'weight': {'$gt': 1}})
>>> for item in cursor:
...   print(item)
...
{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c4617c762b2683dbe52b19'), 'weight': 2}
>>> cursor = mongoose_collection.find({'weight': {'$gt': 1}})
>>> mongooses = list(cursor)
>>> len(mongooses)
1
>>> for item in mongooses:
...   print(item)
...
{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c4617c762b2683dbe52b19'), 'weight': 2}
>>> for item in mongooses:
...   print(item)
...
{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c4617c762b2683dbe52b19'), 'weight': 2}
>>> cursor = list(mongoose_collection.find({'weight': {'$gt': 1}}))
>>> cursor
[{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c4617c762b2683dbe52b19'), 'weight': 2}]
>>> mongoose_collection.find_one()
{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c4617c762b2683dbe52b19'), 'weight': 2}
>>> mongoose_collection.find_one({"eats":'Termites'})
{'name': 'Yellow mongoose', 'eats': 'Termites', '_id': ObjectId('69c4617c762b2683dbe52b1a')}
>>> x = mongoose_collection.find_one({"eats":'Termites'})
>>> x
{'name': 'Yellow mongoose', 'eats': 'Termites', '_id': ObjectId('69c4617c762b2683dbe52b1a')}
>>> x["_id"]
ObjectId('69c4617c762b2683dbe52b1a')
>>> x._id
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'dict' object has no attribute '_id'
>>> str(x["_id"])
'69c4617c762b2683dbe52b1a'
>>> s=str(x["_id"])
>>> s
'69c4617c762b2683dbe52b1a'
>>> ObjectId(s)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
NameError: name 'ObjectId' is not defined. Did you mean: 'object'?
>>> from bson import objectid.ObjectId
  File "<stdin>", line 1
    from bson import objectid.ObjectId
                             ^
SyntaxError: invalid syntax
>>> from bson.objectid import ObjectId
>>> ObjectId(s)
ObjectId('69c4617c762b2683dbe52b1a')
>>> mongoose_collection.find_one({'name': 'Meercat'})
{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c4617c762b2683dbe52b19'), 'weight': 2}
>>> mongoose_collection.delete_one({'name': 'Meercat'})
<mongita.results.DeleteResult object at 0x7e62c3e71a30>
>>> mongoose_collection.find_one({'name': 'Meercat'})
>>> mongoose_collection.insert_one({{'name': 'Meercat', 'does_not_eat': 'Snakes'})
  File "<stdin>", line 1
    mongoose_collection.insert_one({{'name': 'Meercat', 'does_not_eat': 'Snakes'})
                                                                                 ^
SyntaxError: closing parenthesis ')' does not match opening parenthesis '{'
>>> mongoose_collection.insert_one({'name': 'Meercat', 'does_not_eat': 'Snakes'})
<mongita.results.InsertOneResult object at 0x7e62c315ffb0>
>>> mongoose_collection.find_one({'name': 'Meercat'})
{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c464ad762b2683dbe52b1d')}
>>> mongoose_collection.find_one({'weight': {'$gt': 1}})
>>> mongoose_collection.update_one({'name': 'Meercat'}, {'$set': {"weight": 2}})
<mongita.results.UpdateResult object at 0x7e62c315fe90>
>>> mongoose_collection.find_one({'weight': {'$gt': 1}})
{'name': 'Meercat', 'does_not_eat': 'Snakes', '_id': ObjectId('69c464ad762b2683dbe52b1d'), 'weight': 2}
>>> mongoose_collection.find_one({'weight': {'$gt': 300}})
>>> print(mongoose_collection.find_one({'weight': {'$gt': 300}}))
None
>>> print(mongoose_collection.find({'weight': {'$gt': 300}}))
<mongita.cursor.Cursor object at 0x7e62c315ff50>
>>> list(print(mongoose_collection.find({'weight': {'$gt': 300}})))
<mongita.cursor.Cursor object at 0x7e62c315fe90>
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: 'NoneType' object is not iterable
>>>

"""Create the two starting documents without overwriting existing work."""
import argparse
from lesson import DATA_DIR, connect, seed


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--reset", action="store_true",
                        help="delete all documents in the lesson collection and reseed it")
    args = parser.parse_args()
    client, collection = connect()
    try:
        seed(collection, reset=args.reset)
        print("Storage:", DATA_DIR)
        print("Documents:", collection.count_documents({}))
    finally:
        client.close()


if __name__ == "__main__":
    main()

"""Exercise actual Mongita disk storage, isolated from the lesson data."""
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from bson.objectid import ObjectId
from lesson import connect, seed


class MongitaLessonTests(unittest.TestCase):
    def setUp(self):
        self.folder = tempfile.TemporaryDirectory()
        self.client, self.collection = connect(Path(self.folder.name) / "data" / "mongita")
        seed(self.collection)

    def tearDown(self):
        self.client.close()
        self.folder.cleanup()

    def test_seed_preserves_work_and_reset_is_scoped(self):
        self.collection.insert_one({"name": "Student example"})
        other = self.client.hello_world_db.other_collection
        other.insert_one({"keep": True})
        seed(self.collection)
        self.assertEqual(self.collection.count_documents({}), 3)
        seed(self.collection, reset=True)
        self.assertEqual(self.collection.count_documents({}), 2)
        self.assertEqual(other.count_documents({}), 1)

    def test_update_filter_cursor_and_dictionary(self):
        result = self.collection.update_one({"name": "Meercat"}, {"$set": {"weight": 2}})
        self.assertEqual((result.matched_count, result.modified_count), (1, 1))
        cursor = self.collection.find({"weight": {"$gt": 1}})
        saved = list(cursor)
        self.assertEqual([x["name"] for x in saved], ["Meercat"])
        self.assertEqual(list(cursor), [])
        self.assertEqual(len(saved), 1)
        self.assertEqual(saved[0]["does_not_eat"], "Snakes")
        self.assertNotIn("weight", self.collection.find_one({"name": "Yellow mongoose"}))
        self.assertIsNone(self.collection.find_one({"weight": {"$gt": 300}}))
        self.assertEqual(list(self.collection.find({"weight": {"$gt": 300}})), [])
        self.assertIn("_id", list(saved[0]))  # Iterating a dictionary yields keys.

    def test_identifier_delete_reinsert(self):
        old = self.collection.find_one({"name": "Meercat"})
        identifier = str(old["_id"])
        self.assertEqual(self.collection.find_one({"_id": ObjectId(identifier)}), old)
        self.collection.update_one({"_id": old["_id"]}, {"$set": {"weight": 2}})
        self.assertEqual(self.collection.delete_one({"_id": old["_id"]}).deleted_count, 1)
        self.assertIsNone(self.collection.find_one({"_id": old["_id"]}))
        new = self.collection.insert_one({"name": "Meercat", "does_not_eat": "Snakes"})
        self.assertNotEqual(new.inserted_id, old["_id"])
        self.assertEqual(list(self.collection.find({"weight": {"$gt": 1}})), [])

    def test_persistence_in_a_new_process(self):
        self.collection.update_one({"name": "Meercat"}, {"$set": {"weight": 7}})
        self.client.close()
        result = subprocess.run([sys.executable, "-c",
            "from lesson import connect; import sys; "
            "c, col = connect(sys.argv[1]); "
            "print(col.find_one({'name': 'Meercat'})['weight']); c.close()",
            str(Path(self.folder.name) / "data" / "mongita")], cwd=Path(__file__).parent, text=True,
            capture_output=True, check=True)
        self.assertEqual(result.stdout.strip(), "7")


if __name__ == "__main__":
    unittest.main()