code: def test_init_and_checkpoint(temp_workdir):
      with open("keepsake.yaml", "w") as f:
          f.write("repository: file://.keepsake/")
  
      with open("train.py", "w") as fh:
          fh.write("print(1 + 1)")
  
      with open("README.md", "w") as fh:
          fh.write("Hello")
  
      # basic experiment
      experiment = keepsake.init(
          path=".",
          params={"learning_rate": 0.002},
          disable_heartbeat=True,
      )
  
      experiment_tar_path = ".keepsake/experiments/{}.tar.gz".format(
          experiment.id
      )
      wait(
          lambda: os.path.exists(experiment_tar_path),
          timeout_seconds=5,
          sleep_seconds=0.01,
      )
      time.sleep(0.1)  # wait for file to be written
  
      assert len(experiment.id) == 64
      with open(
          ".keepsake/metadata/experiments/{}.json".format(experiment.id)
      ) as fh:
          metadata = json.load(fh)
      assert metadata["id"] == experiment.id
      assert metadata["params"] == {"learning_rate": 0.002}
      assert metadata["host"] == ""
      assert metadata["user"] != ""
      # FIXME: this is broken https://github.com/replicate/keepsake/issues/492
      assert metadata["config"]["repository"].startswith("file://")
      assert metadata["command"] != ""
      assert metadata["path"] == "."
      assert metadata["python_version"] != ""
      assert len(metadata["python_packages"]) > 0
      assert metadata["keepsake_version"] != ""
  
      with tempfile.TemporaryDirectory() as tmpdir:
          with tarfile.open(experiment_tar_path) as tar:
              tar.extractall(tmpdir)
  
          assert (
              open(
                  os.path.join(tmpdir, experiment.id, "train.py")
              ).read()
              == "print(1 + 1)"
          )
          assert os.path.exists(
              os.path.join(tmpdir, experiment.id, "README.md")
          )
  
      # checkpoint with a file
      with open("weights", "w") as fh:
          fh.write("1.2kg")
  
      checkpoint = experiment.checkpoint(
          path="weights", step=1, metrics={"validation_loss": 0.123}
      )
  
      checkpoint_tar_path = ".keepsake/checkpoints/{}.tar.gz".format(
          checkpoint.id
      )
      wait(
          lambda: os.path.exists(checkpoint_tar_path),
          timeout_seconds=5,
          sleep_seconds=0.01,
      )
      time.sleep(0.1)  # wait for file to be written
  
      assert len(checkpoint.id) == 64
      with open(
          ".keepsake/metadata/experiments/{}.json".format(experiment.id)
      ) as fh:
          metadata = json.load(fh)
      assert len(metadata["checkpoints"]) == 1
      checkpoint_metadata = metadata["checkpoints"][0]
      assert checkpoint_metadata["id"] == checkpoint.id
      assert checkpoint_metadata["step"] == 1
      assert checkpoint_metadata["metrics"] == {
          "validation_loss": 0.123
      }
  
      with tempfile.TemporaryDirectory() as tmpdir:
          with tarfile.open(checkpoint_tar_path) as tar:
              tar.extractall(tmpdir)
  
          assert (
              open(
                  os.path.join(tmpdir, checkpoint.id, "weights")
              ).read()
              == "1.2kg"
          )
          assert not os.path.exists(
              os.path.join(tmpdir, checkpoint.id, "train.py")
          )
  
      # checkpoint with a directory
      os.mkdir("data")
      with open("data/weights", "w") as fh:
          fh.write("1.3kg")
  
      checkpoint = experiment.checkpoint(
          path="data", step=1, metrics={"validation_loss": 0.123}
      )
  
      checkpoint_tar_path = ".keepsake/checkpoints/{}.tar.gz".format(
          checkpoint.id
      )
      wait(
          lambda: os.path.exists(checkpoint_tar_path),
          timeout_seconds=5,
          sleep_seconds=0.01,
      )
      time.sleep(0.1)  # wait for file to be written
  
      with tempfile.TemporaryDirectory() as tmpdir:
          with tarfile.open(checkpoint_tar_path) as tar:
              tar.extractall(tmpdir)
  
          assert (
              open(
                  os.path.join(tmpdir, checkpoint.id, "data/weights")
              ).read()
              == "1.3kg"
          )
          assert not os.path.exists(
              os.path.join(tmpdir, checkpoint.id, "train.py")
          )
  
      # checkpoint with no path
      checkpoint = experiment.checkpoint(
          path=None, step=1, metrics={"validation_loss": 0.123}
      )
  
      # wait in case async process tries to create a path anyway
      time.sleep(0.5)
  
      with open(
          ".keepsake/metadata/experiments/{}.json".format(experiment.id)
      ) as fh:
          metadata = json.load(fh)
      assert metadata["checkpoints"][-1]["id"] == checkpoint.id
      assert not os.path.exists(
          ".keepsake/checkpoints/{}.tar.gz".format(checkpoint.id)
      )
  
      # experiment with file
      experiment = keepsake.init(
          path="train.py",
          params={"learning_rate": 0.002},
          disable_heartbeat=True,
      )
  
      experiment_tar_path = ".keepsake/experiments/{}.tar.gz".format(
          experiment.id
      )
      wait(
          lambda: os.path.exists(experiment_tar_path),
          timeout_seconds=5,
          sleep_seconds=0.01,
      )
      time.sleep(0.1)  # wait for file to be written
  
      with tempfile.TemporaryDirectory() as tmpdir:
          with tarfile.open(experiment_tar_path) as tar:
              tar.extractall(tmpdir)
  
          assert (
              open(
                  os.path.join(tmpdir, experiment.id, "train.py")
              ).read()
              == "print(1 + 1)"
          )
          assert not os.path.exists(
              os.path.join(tmpdir, experiment.id, "README.md")
          )
  
      # experiment with no path!
      experiment = keepsake.init(
          path=None,
          params={"learning_rate": 0.002},
          disable_heartbeat=True,
      )
  
      # wait in case async process tries to create a path anyway
      time.sleep(0.5)
  
      with open(
          ".keepsake/metadata/experiments/{}.json".format(experiment.id)
      ) as fh:
          metadata = json.load(fh)
      assert metadata["id"] == experiment.id
      assert metadata["params"] == {"learning_rate": 0.002}
      assert not os.path.exists(
          ".keepsake/experiments/{}.tar.gz".format(experiment.id)
      )
  
imports: import math
  import datetime
  import json
  import os
  import pytest
  import tarfile
  import tempfile
  import time
  from pathlib import Path
  from unittest.mock import patch
  from waiting import wait
  import keepsake
  from keepsake.exceptions import DoesNotExist, ConfigNotFound, IncompatibleRepositoryVersion
  from keepsake.experiment import Experiment, ExperimentList
  from keepsake.project import Project
  from keepsake.metadata import rfc3339_datetime
  from tests.factories import experiment_factory, checkpoint_factory
label_desc: Use image node_id 1 for calling a global function with example usage: test_init_and_checkpoint(temp_workdir) without return types