import re
import errno
import hashlib
+import datetime
+import ciso8601
import time
import threading
def root_collection(self):
return self
+ def get_properties(self):
+ if self._api_response and self._api_response["properties"]:
+ return self._api_response["properties"]
+ else:
+ return {}
+
+ def get_trash_at(self):
+ if self._api_response and self._api_response["trash_at"]:
+ return ciso8601.parse_datetime(self._api_response["trash_at"])
+ else:
+ return None
+
def stream_name(self):
return "."
`save_new()`.
:properties:
- Additional properties of collection.
+ Additional properties of collection. This value will replace any existing
+ properties of collection.
:storage_classes:
Specify desirable storage classes to be used when writing data to Keep.
if storage_classes and type(storage_classes) is not list:
raise errors.ArgumentError("storage_classes must be list type.")
+ if trash_at and type(trash_at) is not datetime.datetime:
+ raise errors.ArgumentError("trash_at must be datetime type.")
+
body={}
if properties:
body["properties"] = properties
if storage_classes:
body["storage_classes_desired"] = storage_classes
- if storage_classes:
- body["trash_at"] = trash_at
+ if trash_at:
+ t = trash_at.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+ body["trash_at"] = t
if not self.committed():
if not self._has_collection_uuid():
If None, defaults to the current user.
:properties:
- Additional properties of collection.
+ Additional properties of collection. This value will replace any existing
+ properties of collection.
:storage_classes:
Specify desirable storage classes to be used when writing data to Keep.
if storage_classes and type(storage_classes) is not list:
raise errors.ArgumentError("storage_classes must be list type.")
+ if trash_at and type(trash_at) is not datetime.datetime:
+ raise errors.ArgumentError("trash_at must be datetime type.")
+
self._my_block_manager().commit_all()
text = self.manifest_text(strip=False)
body["properties"] = properties
if storage_classes:
body["storage_classes_desired"] = storage_classes
- if storage_classes:
- body["trash_at"] = trash_at
+ if trash_at:
+ t = trash_at.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+ body["trash_at"] = t
self._remember_api_response(self._my_api().collections().create(ensure_unique_name=ensure_unique_name, body=body).execute(num_retries=num_retries))
text = self._api_response["manifest_text"]
import re
import sys
import tempfile
+import datetime
import time
import unittest
self.assertEqual(fn0, c.items()[0][0])
self.assertEqual(fn1, c.items()[1][0])
+ def test_get_properties(self):
+ c = Collection()
+ self.assertEqual(c.get_properties(), {})
+ c.save_new(properties={"foo":"bar"})
+ self.assertEqual(c.get_properties(), {"foo":"bar"})
+
+ def test_get_trash_at(self):
+ c = Collection()
+ self.assertEqual(c.get_trash_at(), None)
+ c.save_new(trash_at=datetime.datetime(2111, 1, 1, 11, 11, 11, 111))
+ self.assertEqual(c.get_trash_at(), datetime.datetime(2111, 1, 1, 11, 11, 11, 111))
+
class CollectionOpenModes(run_test_server.TestCaseWithServers):
c = self.create_count_txt()
c.save(properties={'type' : 'Intermediate'},
storage_classes=['archive'],
- trash_at='2100-01-01T00:00:00.000000000Z')
+ trash_at=datetime.datetime(2111, 1, 1, 11, 11, 11, 111111))
self.assertRegex(
c.manifest_text(),
r"^\. 781e5e245d69b566979b86e28d23f2c7\+10\+A[a-f0-9]{40}@[a-f0-9]{8} 0:10:count\.txt$",)
self.assertEqual(c.api_response()["storage_classes_desired"], ['archive'])
self.assertEqual(c.api_response()["properties"], {'type' : 'Intermediate'})
- self.assertEqual(c.api_response()["trash_at"], '2100-01-01T00:00:00.000000000Z')
+ self.assertEqual(c.api_response()["trash_at"], '2111-01-01T11:11:11.111111000Z')
def test_create_and_save_new(self):
c = self.create_count_txt()
c.save_new(properties={'type' : 'Intermediate'},
storage_classes=['archive'],
- trash_at='2100-01-01T00:00:00.000000000Z')
+ trash_at=datetime.datetime(2111, 1, 1, 11, 11, 11, 111111))
self.assertRegex(
c.manifest_text(),
r"^\. 781e5e245d69b566979b86e28d23f2c7\+10\+A[a-f0-9]{40}@[a-f0-9]{8} 0:10:count\.txt$",)
self.assertEqual(c.api_response()["storage_classes_desired"], ['archive'])
self.assertEqual(c.api_response()["properties"], {'type' : 'Intermediate'})
- self.assertEqual(c.api_response()["trash_at"], '2100-01-01T00:00:00.000000000Z')
+ self.assertEqual(c.api_response()["trash_at"], '2111-01-01T11:11:11.111111000Z')
def test_create_and_save_after_commiting(self):
c = self.create_count_txt()
c.save(properties={'type' : 'Intermediate'},
storage_classes=['hot'],
- trash_at='2100-01-01T00:00:00.000000000Z')
+ trash_at=datetime.datetime(2111, 1, 1, 11, 11, 11, 111111))
c.save(properties={'type' : 'Output'},
storage_classes=['cold'],
- trash_at='2200-02-02T22:22:22.222222222Z')
+ trash_at=datetime.datetime(2222, 2, 2, 22, 22, 22, 222222))
self.assertEqual(c.api_response()["storage_classes_desired"], ['cold'])
self.assertEqual(c.api_response()["properties"], {'type' : 'Output'})
- self.assertEqual(c.api_response()["trash_at"], '2200-02-02T22:22:22.222222222Z')
+ self.assertEqual(c.api_response()["trash_at"], '2222-02-02T22:22:22.222222000Z')
def test_create_diff_apply(self):
c1 = self.create_count_txt()