Repository: IDSIA/sacred
Pull request #686 (2019-10-27 17:29:45)

Google Cloud Storage observer
## Motivation
With the advent of the new [Google AI platform (formerly ML-engine)](https://cloud.google.com/ai-platform/) many users will probably like to run experiments on this platform, but due the containerised nature of the platform, the experimenter cannot make use of the existing FileStorageObserver. In order to prevent fragmentation, e.g. using AWS S3 for data storage, it would be favourable to log experiments directly to Google's own servers.

This PR provides an observer for exactly that. By using the Python API for Google Cloud Storage (GCS), the experiments can be logged directly with no overhead to the user.

## Notes

* The code, tests and documentation is based in most part on the existing S3 observer.
* Due to the lack of mocking in place for GCS, the tests can only be run against live storage buckets. Therefore the environment variable `CLOUD_STORAGE_BUCKET` must be set and point to a valid existing bucket.

wohlert: Thanks for the review. The conclusion on my side of the mocking ordeal is that it would be a mostly moot test to use `mock`, as it does not validate any endpoints. I think the current solution is less than ideal, but can still be validated with a service account.

As a side note, the black test is failing due to reasons unrelated to this PR.
JarnoRFB: @wohlert yes the black failure seems to be due to an upgrade of black. I can fix this once merged.

Regarding testing: you summarized it. We should keep this in mind though, once a better solution appears.
JarnoRFB: Would it be possible to integrate testing into CI with something like a *public* account or bucket?
wohlert: It depends on what you mean by public. The bucket itself can be viewable by anyone, but it will still be required to have an owner. In any case, a [service account](https://cloud.google.com/iam/docs/service-accounts) need to be set up from which the service can validate identity, e.g. from a key file or API token.
JarnoRFB: Ok so nothing that we would want to put in the CI I guess. Then let's just merge this and come back to it when a better alternative for testing appears.

Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 66d9021..cd6b6d5 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,7 +1,7 @@
 fail_fast: true
 repos:
 -   repo: https://github.com/psf/black
-    rev: master
+    rev: stable
     hooks:
     - id: black
       language_version: python3.6
diff --git a/dev-requirements.txt b/dev-requirements.txt
index 6f73e45..f1288da 100644
--- a/dev-requirements.txt
+++ b/dev-requirements.txt
@@ -31,5 +31,6 @@ py-cpuinfo==4.0
 boto3>=1.9.0
 moto>=1.3.13
 google-compute-engine>=2.8.0
+google-cloud-storage==1.20.0
 pre-commit==1.18.0
 boto3>=1.9.0
diff --git a/docs/observers.rst b/docs/observers.rst
index 3d4c1b7..b6052f3 100644
--- a/docs/observers.rst
+++ b/docs/observers.rst
@@ -9,7 +9,7 @@ Observers have a ``priority`` attribute, and are run in order of descending
 priority. The first observer determines the ``_id`` of the run.
 
 
-At the moment there are five observers that are shipped with Sacred:
+At the moment there are seven observers that are shipped with Sacred:
 
  * The main one is the :ref:`mongo_observer` which stores all information in a
    `MongoDB <http://www.mongodb.org/>`_.
@@ -22,6 +22,8 @@ At the moment there are five observers that are shipped with Sacred:
    relevant information there.
  * The :ref:`s3_observer` stores run information in an AWS S3 bucket, within
    a given prefix/directory
+ * The :ref:`gcs_observer` stores run information in a provided Google Cloud
+   Storage bucket, within a given prefix/directory
  * The :ref:`queue_observer` can be used to wrap any of the above observers.
    It will put the processing of observed events on a fault-tolerant 
    queue in a background process. This is useful for observers that rely
@@ -641,6 +643,43 @@ created in ascending order, and each run directory will contain the files specif
 FileStorageObserver Directory Structure documentation above.
 
 
+Google Cloud Storage Observer
+============
+
+.. note::
+    Requires the `google cloud storage <https://cloud.google.com/storage/docs/reference/libraries/>`_ package.
+    Install with ``pip install google-cloud-storage``.
+
+The Google Cloud Storage Observer allows for experiments to be logged into cloud storage buckets
+provided by Google. In order to use this observer, the user must have created a bucket on the service
+prior to the running an experiment using this observer.
+
+
+Adding a GoogleCloudStorageObserver
+--------------------
+
+To create an GoogleCloudStorageObserver in Python:
+
+.. code-block:: python
+
+    from sacred.observers import GoogleCloudStorageObserver
+    ex.observers.append(GoogleCloudStorageObserver(bucket='bucket-name',
+                                                   basedir='/experiment-name/'))
+
+In order for the observer to correctly connect to the provided bucket, The environment variable
+`` GOOGLE_APPLICATION_CREDENTIALS``  needs to be set by the user. This variable should point to a
+valid JSON file containing Google authorisation credentials
+(see: `Google Cloud authentication <https://cloud.google.com/docs/authentication/getting-started/>`_).
+
+Directory Structure
+--------------------
+
+GoogleCloudStorageObserver follow the same conventions as FileStorageObservers when it comes to directory
+structure within a bucket: within ``gs://<bucket>/basedir/`` numeric run directories will be
+created in ascending order, and each run directory will contain the files specified within the
+FileStorageObserver Directory Structure documentation above.
+
+
 Slack Observer
 ==============
 
diff --git a/sacred/experiment.py b/sacred/experiment.py
index 1a20792..dd4204f 100755
--- a/sacred/experiment.py
+++ b/sacred/experiment.py
@@ -558,8 +558,9 @@ class Experiment(Ingredient):
     def _check_command(self, cmd_name):
         commands = dict(self.gather_commands())
         if cmd_name is not None and cmd_name not in commands:
-            return 'Error: Command "{}" not found. Available commands are: ' "{}".format(
-                cmd_name, ", ".join(commands.keys())
+            return (
+                'Error: Command "{}" not found. Available commands are: '
+                "{}".format(cmd_name, ", ".join(commands.keys()))
             )
 
         if cmd_name is None:
diff --git a/sacred/observers/__init__.py b/sacred/observers/__init__.py
index c1cfe5d..c55f527 100644
--- a/sacred/observers/__init__.py
+++ b/sacred/observers/__init__.py
@@ -7,6 +7,7 @@ from sacred.observers.slack import SlackObserver
 from sacred.observers.telegram_obs import TelegramObserver
 from sacred.observers.s3_observer import S3Observer
 from sacred.observers.queue import QueueObserver
+from sacred.observers.gcs_observer import GoogleCloudStorageObserver
 
 
 __all__ = (
@@ -21,4 +22,5 @@ __all__ = (
     "TelegramObserver",
     "S3Observer",
     "QueueObserver",
+    "GoogleCloudStorageObserver",
 )
diff --git a/sacred/observers/gcs_observer.py b/sacred/observers/gcs_observer.py
new file mode 100644
index 0000000..8e063f2
--- /dev/null
+++ b/sacred/observers/gcs_observer.py
@@ -0,0 +1,346 @@
+import json
+import os
+import os.path
+import re
+from typing import Optional
+
+from sacred.commandline_options import cli_option
+from sacred.dependencies import get_digest
+from sacred.observers.base import RunObserver
+from sacred.serializer import flatten
+from sacred.utils import PathType
+
+DEFAULT_GCS_PRIORITY = 20
+
+
+def _is_valid_bucket(bucket_name: str):
+    """Validates correctness of bucket naming.
+
+    Reference: https://cloud.google.com/storage/docs/naming
+    """
+    if bucket_name.startswith("gs://"):
+        return False
+
+    if len(bucket_name) < 3 or len(bucket_name) > 63:
+        return False
+
+    # IP address
+    if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", bucket_name):
+        return False
+
+    if not re.fullmatch(r"([^A-Z]|-|_|[.]|)+", bucket_name):
+        return False
+
+    if ".." in bucket_name:
+        return False
+
+    if "goog" in bucket_name or "g00g" in bucket_name:
+        return False
+
+    return True
+
+
+def gcs_join(*args):
+    return "/".join(args)
+
+
+class GoogleCloudStorageObserver(RunObserver):
+    VERSION = "GoogleCloudStorageObserver-0.1.0"
+
+    def __init__(
+        self,
+        bucket: str,
+        basedir: PathType,
+        resource_dir: Optional[PathType] = None,
+        source_dir: Optional[PathType] = None,
+        priority: Optional[int] = DEFAULT_GCS_PRIORITY,
+    ):
+        """Constructor for a GoogleCloudStorageObserver object.
+
+        Run when the object is first created,
+        before it's used within an experiment.
+
+        Parameters
+        ----------
+        bucket
+            The name of the bucket you want to store results in.
+            Needs to be a valid bucket name without 'gs://'
+        basedir
+            The relative path inside your bucket where you want this experiment to store results
+        resource_dir
+            Where to store resources for this experiment. By
+            default, will be <basedir>/_resources
+        source_dir
+            Where to store code sources for this experiment. By
+            default, will be <basedir>/sources
+        priority
+            The priority to assign to this observer if
+            multiple observers are present
+        """
+        if not _is_valid_bucket(bucket):
+            raise ValueError(
+                "Your chosen bucket name doesn't follow Google Cloud Storage bucket naming rules"
+            )
+        resource_dir = resource_dir or "/".join([basedir, "_resources"])
+        source_dir = source_dir or "/".join([basedir, "_sources"])
+
+        self.basedir = basedir
+        self.bucket_id = bucket
+
+        self.resource_dir = resource_dir
+        self.source_dir = source_dir
+        self.priority = priority
+        self.dir = None
+        self.run_entry = None
+        self.config = None
+        self.info = None
+        self.cout = ""
+        self.cout_write_cursor = 0
+        self.saved_metrics = {}
+
+        from google.cloud import storage
+        import google.auth.exceptions
+
+        try:
+            client = storage.Client()
+        except google.auth.exceptions.DefaultCredentialsError:
+            raise ConnectionError(
+                "Could not create Google Cloud Storage observer, are you "
+                "sure that you have set environment variable GOOGLE_APPLICATION_CREDENTIALS?"
+            )
+
+        self.bucket = client.bucket(bucket)
+
+    def _objects_exist_in_dir(self, prefix):
+        # This should be run after you've confirmed the bucket
+        # exists, and will error out if it does not exist
+        all_blobs = [blob for blob in self.bucket.list_blobs(prefix=prefix)]
+        return len(all_blobs) > 0
+
+    def _list_gcs_subdirs(self, prefix=None):
+        if prefix is None:
+            prefix = self.basedir
+
+        iterator = self.bucket.list_blobs(prefix=prefix, delimiter="/")
+        prefixes = set()
+        for page in iterator.pages:
+            prefixes.update(page.prefixes)
+
+        return list(prefixes)
+
+    def _determine_run_dir(self, _id):
+        if _id is None:
+            basepath = os.path.join(self.basedir, "")
+            bucket_path_subdirs = self._list_gcs_subdirs(prefix=basepath)
+
+            if not bucket_path_subdirs:
+                max_run_id = 0
+            else:
+                relative_paths = [
+                    path.replace(self.basedir, "").strip("/")
+                    for path in bucket_path_subdirs
+                ]
+                integer_directories = [int(d) for d in relative_paths if d.isdigit()]
+                if not integer_directories:
+                    max_run_id = 0
+                else:
+                    # If there are directories under basedir that aren't
+                    # numeric run directories, ignore those
+                    max_run_id = max(integer_directories)
+
+            _id = max_run_id + 1
+
+        self.dir = gcs_join(self.basedir, str(_id))
+        if self._objects_exist_in_dir(self.dir):
+            raise FileExistsError("GCS dir at {} already exists".format(self.dir))
+        return _id
+
+    def queued_event(
+        self, ex_info, command, host_info, queue_time, config, meta_info, _id
+    ):
+        _id = self._determine_run_dir(_id)
+
+        self.run_entry = {
+            "experiment": dict(ex_info),
+            "command": command,
+            "host": dict(host_info),
+            "meta": meta_info,
+            "status": "QUEUED",
+        }
+        self.config = config
+        self.info = {}
+
+        self.save_json(self.run_entry, "run.json")
+        self.save_json(self.config, "config.json")
+
+        for s, m in ex_info["sources"]:
+            self.save_file(s)
+
+        return _id
+
+    def save_sources(self, ex_info):
+        base_dir = ex_info["base_dir"]
+        source_info = []
+        for s, m in ex_info["sources"]:
+            abspath = os.path.join(base_dir, s)
+            store_path, md5sum = self.find_or_save(abspath, self.source_dir)
+            source_info.append([s, os.path.relpath(store_path, self.basedir)])
+        return source_info
+
+    def started_event(
+        self, ex_info, command, host_info, start_time, config, meta_info, _id
+    ):
+
+        _id = self._determine_run_dir(_id)
+
+        ex_info["sources"] = self.save_sources(ex_info)
+
+        self.run_entry = {
+            "experiment": dict(ex_info),
+            "command": command,
+            "host": dict(host_info),
+            "start_time": start_time.isoformat(),
+            "meta": meta_info,
+            "status": "RUNNING",
+            "resources": [],
+            "artifacts": [],
+            "heartbeat": None,
+        }
+        self.config = config
+        self.info = {}
+        self.cout = ""
+        self.cout_write_cursor = 0
+
+        self.save_json(self.run_entry, "run.json")
+        self.save_json(self.config, "config.json")
+        self.save_cout()
+
+        return _id
+
+    def find_or_save(self, filename, store_dir):
+        source_name, ext = os.path.splitext(os.path.basename(filename))
+        md5sum = get_digest(filename)
+        store_name = source_name + "_" + md5sum + ext
+        store_path = gcs_join(store_dir, store_name)
+        if len(self._list_gcs_subdirs(prefix=store_path)) == 0:
+            self.save_file_to_base(filename, store_path)
+        return store_path, md5sum
+
+    def put_data(self, key, binary_data):
+        blob = self.bucket.blob(key)
+        blob.upload_from_file(binary_data)
+
+    def save_json(self, obj, filename):
+        key = gcs_join(self.dir, filename)
+        blob = self.bucket.blob(key)
+        blob.upload_from_string(
+            json.dumps(flatten(obj), sort_keys=True, indent=2), content_type="text/json"
+        )
+
+    def save_file(self, filename, target_name=None):
+        target_name = target_name or os.path.basename(filename)
+        key = gcs_join(self.dir, target_name)
+        self.put_data(key, open(filename, "rb"))
+
+    def save_file_to_base(self, filename, target_name=None):
+        target_name = target_name or os.path.basename(filename)
+        self.put_data(target_name, open(filename, "rb"))
+
+    def save_directory(self, source_dir, target_name):
+        target_name = target_name or os.path.basename(source_dir)
+        all_files = []
+        for root, dirs, files in os.walk(source_dir):
+            all_files += [os.path.join(root, f) for f in files]
+
+        for filename in all_files:
+            file_location = gcs_join(
+                self.dir, target_name, os.path.relpath(filename, source_dir)
+            )
+            self.put_data(file_location, open(filename, "rb"))
+
+    def save_cout(self):
+        binary_data = self.cout[self.cout_write_cursor :].encode("utf-8")
+        key = gcs_join(self.dir, "cout.txt")
+        blob = self.bucket.blob(key)
+        blob.upload_from_string(binary_data, content_type="text/plain")
+        self.cout_write_cursor = len(self.cout)
+
+    def heartbeat_event(self, info, captured_out, beat_time, result):
+        self.info = info
+        self.run_entry["heartbeat"] = beat_time.isoformat()
+        self.run_entry["result"] = result
+        self.cout = captured_out
+        self.save_cout()
+        self.save_json(self.run_entry, "run.json")
+        if self.info:
+            self.save_json(self.info, "info.json")
+
+    def completed_event(self, stop_time, result):
+        self.run_entry["stop_time"] = stop_time.isoformat()
+        self.run_entry["result"] = result
+        self.run_entry["status"] = "COMPLETED"
+
+        self.save_json(self.run_entry, "run.json")
+
+    def interrupted_event(self, interrupt_time, status):
+        self.run_entry["stop_time"] = interrupt_time.isoformat()
+        self.run_entry["status"] = status
+        self.save_json(self.run_entry, "run.json")
+
+    def failed_event(self, fail_time, fail_trace):
+        self.run_entry["stop_time"] = fail_time.isoformat()
+        self.run_entry["status"] = "FAILED"
+        self.run_entry["fail_trace"] = fail_trace
+        self.save_json(self.run_entry, "run.json")
+
+    def resource_event(self, filename):
+        store_path, md5sum = self.find_or_save(filename, self.resource_dir)
+        self.run_entry["resources"].append([filename, store_path])
+        self.save_json(self.run_entry, "run.json")
+
+    def artifact_event(self, name, filename, metadata=None, content_type=None):
+        self.save_file(filename, name)
+        self.run_entry["artifacts"].append(name)
+        self.save_json(self.run_entry, "run.json")
+
+    def log_metrics(self, metrics_by_name, info):
+        """Store new measurements into metrics.json."""
+        for metric_name, metric_ptr in metrics_by_name.items():
+
+            if metric_name not in self.saved_metrics:
+                self.saved_metrics[metric_name] = {
+                    "values": [],
+                    "steps": [],
+                    "timestamps": [],
+                }
+
+            self.saved_metrics[metric_name]["values"] += metric_ptr["values"]
+            self.saved_metrics[metric_name]["steps"] += metric_ptr["steps"]
+
+            timestamps_norm = [ts.isoformat() for ts in metric_ptr["timestamps"]]
+            self.saved_metrics[metric_name]["timestamps"] += timestamps_norm
+
+        self.save_json(self.saved_metrics, "metrics.json")
+
+    def __eq__(self, other):
+        if isinstance(other, GoogleCloudStorageObserver):
+            return self.bucket_id == other.bucket_id and self.basedir == other.basedir
+        else:
+            return False
+
+
+@cli_option("-G", "--gcs")
+def gcs_option(args, run):
+    """Add a Google Cloud Storage File observer to the experiment.
+
+    The argument value should be `gs://<bucket>/path/to/exp`.
+    """
+    match_obj = re.match(r"gs:\/\/([^\/]*)\/(.*)", args)
+    if match_obj is None or len(match_obj.groups()) != 2:
+        raise ValueError(
+            "Valid bucket specification not found. "
+            "Enter bucket and directory path like: "
+            "gs://<bucket>/path/to/exp"
+        )
+    bucket, basedir = match_obj.groups()
+    run.observers.append(GoogleCloudStorageObserver(bucket=bucket, basedir=basedir))
diff --git a/tox.ini b/tox.ini
index 023c55b..1301c36 100644
--- a/tox.ini
+++ b/tox.ini
@@ -79,7 +79,8 @@ commands =
 
 [testenv:black]
 basepython = python
-deps = git+https://github.com/psf/black
+deps = 
+    black==19.10b0
 commands = 
     black --check sacred/ tests/
 


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 1131c09..cc5af73 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,6 +1,8 @@
 #!/usr/bin/env python
 # coding=utf-8
-
+import pytest
+import tempfile
+import hashlib
 import os.path
 import re
 import shlex
@@ -82,5 +84,26 @@ def pytest_addoption(parser):
     )
 
 
+@pytest.fixture
+def tmpfile():
+    # NOTE: instead of using a with block and delete=True we are creating and
+    # manually deleting the file, such that we can close it before running the
+    # tests. This is necessary since on Windows we can not open the same file
+    # twice, so for the FileStorageObserver to read it, we need to close it.
+    f = tempfile.NamedTemporaryFile(suffix=".py", delete=False)
+
+    f.content = "import sacred\n"
+    f.write(f.content.encode())
+    f.flush()
+    f.seek(0)
+    f.md5sum = hashlib.md5(f.read()).hexdigest()
+
+    f.close()
+
+    yield f
+
+    os.remove(f.name)
+
+
 # Deactivate GPU info to speed up tests
 SETTINGS.HOST_INFO.INCLUDE_GPU_INFO = False
diff --git a/tests/test_observers/test_file_storage_observer.py b/tests/test_observers/test_file_storage_observer.py
index de4d243..7a03b49 100644
--- a/tests/test_observers/test_file_storage_observer.py
+++ b/tests/test_observers/test_file_storage_observer.py
@@ -2,9 +2,7 @@
 # coding=utf-8
 
 import datetime
-import hashlib
 import os
-import tempfile
 from copy import copy
 import pytest
 import json
@@ -42,27 +40,6 @@ def dir_obs(tmpdir):
     return basedir, FileStorageObserver(basedir.strpath)
 
 
-@pytest.fixture
-def tmpfile():
-    # NOTE: instead of using a with block and delete=True we are creating and
-    # manually deleting the file, such that we can close it before running the
-    # tests. This is necessary since on Windows we can not open the same file
-    # twice, so for the FileStorageObserver to read it, we need to close it.
-    f = tempfile.NamedTemporaryFile(suffix=".py", delete=False)
-
-    f.content = "import sacred\n"
-    f.write(f.content.encode())
-    f.flush()
-    f.seek(0)
-    f.md5sum = hashlib.md5(f.read()).hexdigest()
-
-    f.close()
-
-    yield f
-
-    os.remove(f.name)
-
-
 def test_fs_observer_create_does_not_create_basedir(dir_obs):
     basedir, obs = dir_obs
     assert not basedir.exists()
diff --git a/tests/test_observers/test_gcs_observer.py b/tests/test_observers/test_gcs_observer.py
new file mode 100644
index 0000000..845f66c
--- /dev/null
+++ b/tests/test_observers/test_gcs_observer.py
@@ -0,0 +1,240 @@
+import pytest
+import datetime
+import os
+import json
+
+from sacred.observers import GoogleCloudStorageObserver
+
+storage = pytest.importorskip("google.cloud.storage")
+
+T1 = datetime.datetime(1999, 5, 4, 3, 2, 1, 0)
+T2 = datetime.datetime(1999, 5, 5, 5, 5, 5, 5)
+
+try:
+    BUCKET = os.environ["CLOUD_STORAGE_BUCKET"]
+    _CREDENTIALS = os.environ["GOOGLE_APPLICATION_CREDENTIALS"]
+except KeyError as key:
+    print("Skipping test due to missing environment variable", key)
+    pytest.skip("skipping google authentication-only tests", allow_module_level=True)
+
+BASEDIR = "sacred-tests"
+
+
+def gcs_join(*args):
+    return "/".join(args)
+
+
+def _delete_bucket_directory(bucket, basedir):
+    blobs = bucket.list_blobs(prefix=basedir)
+    for blob in blobs:
+        blob.delete()
+
+
+@pytest.fixture()
+def sample_run():
+    exp = {"name": "test_exp", "sources": [], "doc": "", "base_dir": "/tmp"}
+    host = {"hostname": "test_host", "cpu_count": 1, "python_version": "3.4"}
+    config = {"config": "True", "foo": "bar", "answer": 42}
+    command = "run"
+    meta_info = {"comment": "test run"}
+    return {
+        "_id": None,
+        "ex_info": exp,
+        "command": command,
+        "host_info": host,
+        "start_time": T1,
+        "config": config,
+        "meta_info": meta_info,
+    }
+
+
+@pytest.fixture
+def observer(basedir=None):
+    if basedir is None:
+        basedir = BASEDIR
+
+    _observer = GoogleCloudStorageObserver(bucket=BUCKET, basedir=basedir)
+
+    yield _observer
+
+    # Make sure to delete directory afterwards
+    _delete_bucket_directory(_observer.bucket, basedir)
+
+
+def _get_blob(bucket, directory, filename):
+    prefixed_blobs = [blob for blob in bucket.list_blobs(prefix=directory)]
+    file_blob = next((blob for blob in prefixed_blobs if filename in blob.name))
+    return file_blob
+
+
+def _bucket_exists(bucket):
+    all_blobs = [blob for blob in bucket.list_blobs()]
+    return len(all_blobs) > 0
+
+
+def _file_exists(bucket, directory, filename):
+    file_blob = _get_blob(bucket, directory, filename)
+    return gcs_join(directory, filename) == file_blob.name
+
+
+def _get_file_data(bucket, directory, filename):
+    file_blob = _get_blob(bucket, directory, filename)
+    return file_blob.download_as_string()
+
+
+def test_gcs_observer_started_event_creates_bucket(observer, sample_run):
+    bucket = observer.bucket
+    _id = observer.started_event(**sample_run)
+    run_dir = gcs_join(BASEDIR, str(_id))
+
+    assert _bucket_exists(bucket)
+    assert _file_exists(bucket, run_dir, filename="cout.txt")
+    assert _file_exists(bucket, run_dir, filename="config.json")
+    assert _file_exists(bucket, run_dir, filename="run.json")
+
+    config = _get_file_data(bucket, run_dir, filename="config.json")
+    assert json.loads(config.decode("utf-8")) == sample_run["config"]
+
+    run = _get_file_data(bucket, run_dir, filename="run.json")
+    assert json.loads(run.decode("utf-8")) == {
+        "experiment": sample_run["ex_info"],
+        "command": sample_run["command"],
+        "host": sample_run["host_info"],
+        "start_time": T1.isoformat(),
+        "heartbeat": None,
+        "meta": sample_run["meta_info"],
+        "resources": [],
+        "artifacts": [],
+        "status": "RUNNING",
+    }
+
+
+def test_gcs_observer_started_event_increments_run_id(observer, sample_run):
+    _id = observer.started_event(**sample_run)
+    _id2 = observer.started_event(**sample_run)
+    assert _id + 1 == _id2
+
+
+def test_gcs_observer_equality():
+    obs_one = GoogleCloudStorageObserver(bucket=BUCKET, basedir=BASEDIR)
+    obs_two = GoogleCloudStorageObserver(bucket=BUCKET, basedir=BASEDIR)
+    assert obs_one == obs_two
+
+    test_directory = "sacred-tests-2/dir"
+    different_basedir = GoogleCloudStorageObserver(
+        bucket=BUCKET, basedir=test_directory
+    )
+    assert obs_one != different_basedir
+
+    _delete_bucket_directory(obs_one.bucket, BASEDIR)
+    _delete_bucket_directory(different_basedir.bucket, test_directory)
+
+
+def test_raises_error_on_duplicate_id_directory(observer, sample_run):
+    observer.started_event(**sample_run)
+    sample_run["_id"] = 1
+    with pytest.raises(FileExistsError):
+        observer.started_event(**sample_run)
+
+
+def test_completed_event_updates_run_json(observer, sample_run):
+    observer.started_event(**sample_run)
+    run = json.loads(
+        _get_file_data(observer.bucket, observer.dir, filename="run.json").decode(
+            "utf-8"
+        )
+    )
+    assert run["status"] == "RUNNING"
+    observer.completed_event(T2, "success!")
+    run = json.loads(
+        _get_file_data(observer.bucket, observer.dir, filename="run.json").decode(
+            "utf-8"
+        )
+    )
+    assert run["status"] == "COMPLETED"
+
+
+def test_interrupted_event_updates_run_json(observer, sample_run):
+    observer.started_event(**sample_run)
+    run = json.loads(
+        _get_file_data(observer.bucket, observer.dir, filename="run.json").decode(
+            "utf-8"
+        )
+    )
+    assert run["status"] == "RUNNING"
+    observer.interrupted_event(T2, "SERVER_EXPLODED")
+    run = json.loads(
+        _get_file_data(observer.bucket, observer.dir, filename="run.json").decode(
+            "utf-8"
+        )
+    )
+    assert run["status"] == "SERVER_EXPLODED"
+
+
+def test_failed_event_updates_run_json(observer, sample_run):
+    observer.started_event(**sample_run)
+    run = json.loads(
+        _get_file_data(observer.bucket, observer.dir, "run.json").decode("utf-8")
+    )
+    assert run["status"] == "RUNNING"
+    observer.failed_event(T2, "Everything imaginable went wrong")
+    run = json.loads(
+        _get_file_data(observer.bucket, observer.dir, "run.json").decode("utf-8")
+    )
+    assert run["status"] == "FAILED"
+
+
+def test_queued_event_updates_run_json(observer, sample_run):
+    del sample_run["start_time"]
+    sample_run["queue_time"] = T2
+    observer.queued_event(**sample_run)
+    run = json.loads(
+        _get_file_data(observer.bucket, observer.dir, "run.json").decode("utf-8")
+    )
+    assert run["status"] == "QUEUED"
+
+
+def test_artifact_event_works(observer, sample_run, tmpfile):
+    observer.started_event(**sample_run)
+    observer.artifact_event("test_artifact.py", tmpfile.name)
+
+    assert _file_exists(observer.bucket, observer.dir, "test_artifact.py")
+    artifact_data = _get_file_data(
+        observer.bucket, observer.dir, "test_artifact.py"
+    ).decode("utf-8")
+    assert artifact_data == tmpfile.content
+
+
+@pytest.fixture
+def valid_buckets():
+    return [
+        "this_bucket_is_valid",
+        "th15_8uck3t_15_v4l1d",
+        "this-bucket-is-valid",
+        "this-bucket.is-valid",
+    ]
+
+
+@pytest.fixture
+def invalid_buckets():
+    return [
+        "hi",
+        "goog-24",
+        "this-bucket..is-invalid",
+        "-this-bucket-is-invalid",
+        "this-BUCKET-is-invalid",
+        "this-google-is-invalid",
+        "this-g00gle-is-invalid",
+        "192.168.5.4",
+    ]
+
+
+def test_does_not_raise_error_on_valid_bucket_name(valid_buckets):
+    for bucket_name in valid_buckets:
+        _ = GoogleCloudStorageObserver(bucket=bucket_name, basedir=BASEDIR)
+
+
+def test_raises_error_on_invalid_bucket_name(invalid_buckets):
+    for bucket_name in invalid_buckets:
+        with pytest.raises(ValueError):
+            _ = GoogleCloudStorageObserver(bucket=bucket_name, basedir=BASEDIR)
diff --git a/tests/test_observers/test_s3_observer.py b/tests/test_observers/test_s3_observer.py
index 8b1d47d..1028ccc 100644
--- a/tests/test_observers/test_s3_observer.py
+++ b/tests/test_observers/test_s3_observer.py
@@ -2,13 +2,10 @@
 # coding=utf-8
 
 import datetime
-import os
 import pytest
 import json
 
 from sacred.observers import S3Observer
-import tempfile
-import hashlib
 
 moto = pytest.importorskip("moto")
 boto3 = pytest.importorskip("boto3")
@@ -49,27 +46,6 @@ def observer():
     return S3Observer(bucket=BUCKET, basedir=BASEDIR, region=REGION)
 
 
-@pytest.fixture
-def tmpfile():
-    # NOTE: instead of using a with block and delete=True we are creating and
-    # manually deleting the file, such that we can close it before running the
-    # tests. This is necessary since on Windows we can not open the same file
-    # twice, so for the FileStorageObserver to read it, we need to close it.
-    f = tempfile.NamedTemporaryFile(suffix=".py", delete=False)
-
-    f.content = "import sacred\n"
-    f.write(f.content.encode())
-    f.flush()
-    f.seek(0)
-    f.md5sum = hashlib.md5(f.read()).hexdigest()
-
-    f.close()
-
-    yield f
-
-    os.remove(f.name)
-
-
 def _bucket_exists(bucket_name):
     from botocore.exceptions import ClientError
 
diff --git a/tests/test_observers/test_sql_observer.py b/tests/test_observers/test_sql_observer.py
index c1417d0..fcec0cd 100644
--- a/tests/test_observers/test_sql_observer.py
+++ b/tests/test_observers/test_sql_observer.py
@@ -3,11 +3,8 @@
 
 
 import datetime
-import hashlib
-import os
 
 import pytest
-import tempfile
 from sacred.serializer import json
 
 sqlalchemy = pytest.importorskip("sqlalchemy")
@@ -74,27 +71,6 @@ def sample_run():
     }
 
 
-@pytest.fixture
-def tmpfile():
-    # NOTE: instead of using a with block and delete=True we are creating and
-    # manually deleting the file, such that we can close it before running the
-    # tests. This is necessary since on Windows we can not open the same file
-    # twice, so for the FileStorageObserver to read it, we need to close it.
-    f = tempfile.NamedTemporaryFile(suffix=".py", delete=False)
-
-    f.content = "import sacred\n"
-    f.write(f.content.encode())
-    f.flush()
-    f.seek(0)
-    f.md5sum = hashlib.md5(f.read()).hexdigest()
-
-    f.close()
-
-    yield f
-
-    os.remove(f.name)
-
-
 def test_sql_observer_started_event_creates_run(sql_obs, sample_run, session):
     sample_run["_id"] = None
     _id = sql_obs.started_event(**sample_run)