�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fopt%2F'>opt</a>/<a href='?path=%2Fopt%2Fgsutil%2F'>gsutil</a>/<a href='?path=%2Fopt%2Fgsutil%2Fgslib%2F'>gslib</a>/<a href='?path=%2Fopt%2Fgsutil%2Fgslib%2Ftests%2F'>tests</a>/</p><hr><h3>✏️ Editing: test_lifecycle.py</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'># -*- coding: utf-8 -*-
# Copyright 2013 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the &quot;License&quot;);
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an &quot;AS IS&quot; BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
&quot;&quot;&quot;Integration tests for lifecycle command.&quot;&quot;&quot;

from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals

import json
import os
import posixpath
from unittest import mock
from xml.dom.minidom import parseString

from gslib.cs_api_map import ApiSelector
import gslib.tests.testcase as testcase
from gslib.tests.testcase.integration_testcase import SkipForS3
from gslib.tests.util import ObjectToURI as suri
from gslib.tests.util import SetBotoConfigForTest
from gslib.tests.util import SetEnvironmentForTest
from gslib.tests.util import unittest
from gslib.utils.retry_util import Retry
from gslib.utils.translation_helper import LifecycleTranslation
from gslib.utils import shim_util


@SkipForS3(&#039;Lifecycle command is only supported for gs:// URLs&#039;)
class TestSetLifecycle(testcase.GsUtilIntegrationTestCase):
  &quot;&quot;&quot;Integration tests for lifecycle command.&quot;&quot;&quot;

  empty_doc1 = &#039;{}&#039;

  xml_doc = parseString(
      &#039;&lt;LifecycleConfiguration&gt;&lt;Rule&gt;&#039;
      &#039;&lt;Action&gt;&lt;Delete/&gt;&lt;/Action&gt;&#039;
      &#039;&lt;Condition&gt;&lt;Age&gt;365&lt;/Age&gt;&lt;/Condition&gt;&#039;
      &#039;&lt;/Rule&gt;&lt;/LifecycleConfiguration&gt;&#039;).toprettyxml(indent=&#039;    &#039;)

  bad_doc = (
      &#039;{&quot;rule&quot;: [{&quot;action&quot;: {&quot;type&quot;: &quot;Add&quot;}, &quot;condition&quot;: {&quot;age&quot;: 365}}]}\n&#039;)

  lifecycle_doc = (
      &#039;{&quot;rule&quot;: [&#039;
      &#039;{&quot;action&quot;: {&quot;type&quot;: &quot;Delete&quot;}, &quot;condition&quot;: {&quot;age&quot;: 365}}, &#039;
      &#039;{&quot;action&quot;: {&quot;type&quot;: &quot;SetStorageClass&quot;, &quot;storageClass&quot;: &quot;NEARLINE&quot;},&#039;
      &#039; &quot;condition&quot;: {&quot;matchesStorageClass&quot;: [&quot;STANDARD&quot;], &quot;age&quot;: 366}}]}\n&#039;)
  lifecycle_json_obj = json.loads(lifecycle_doc)

  lifecycle_doc_bucket_style = (&#039;{&quot;lifecycle&quot;: &#039; + lifecycle_doc.rstrip() +
                                &#039;}\n&#039;)

  # TODO: Remove once Boto is updated to support new fields.
  lifecycle_doc_without_storage_class_fields = (
      &#039;{&quot;rule&quot;: [{&quot;action&quot;: {&quot;type&quot;: &quot;Delete&quot;}, &quot;condition&quot;: {&quot;age&quot;: 365}}]}\n&#039;)

  lifecycle_created_before_doc = (
      &#039;{&quot;rule&quot;: [{&quot;action&quot;: {&quot;type&quot;: &quot;Delete&quot;}, &quot;condition&quot;: &#039;
      &#039;{&quot;createdBefore&quot;: &quot;2014-10-01&quot;}}]}\n&#039;)
  lifecycle_created_before_json_obj = json.loads(lifecycle_created_before_doc)

  lifecycle_with_falsy_condition_values = (
      &#039;{&quot;rule&quot;: [{&quot;action&quot;: {&quot;type&quot;: &quot;Delete&quot;}, &quot;condition&quot;: {&#039;
      &#039;&quot;age&quot;: 0, &quot;isLive&quot;: false, &quot;numNewerVersions&quot;: 0}}]}&#039;)

  no_lifecycle_config = &#039;has no lifecycle configuration.&#039;

  def test_lifecycle_translation(self):
    &quot;&quot;&quot;Tests lifecycle translation for various formats.&quot;&quot;&quot;
    # TODO: Use lifecycle_doc again once Boto is updated to support new fields.
    # json_text = self.lifecycle_doc
    json_text = self.lifecycle_doc_without_storage_class_fields

    entries_list = LifecycleTranslation.JsonLifecycleToMessage(json_text)
    boto_lifecycle = LifecycleTranslation.BotoLifecycleFromMessage(entries_list)
    converted_entries_list = LifecycleTranslation.BotoLifecycleToMessage(
        boto_lifecycle)
    converted_json_text = LifecycleTranslation.JsonLifecycleFromMessage(
        converted_entries_list)
    self.assertEqual(json.loads(json_text), json.loads(converted_json_text))

  def test_default_lifecycle(self):
    bucket_uri = self.CreateBucket()
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)
    self.assertIn(self.no_lifecycle_config, stdout)

  def test_set_empty_lifecycle1(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(contents=self.empty_doc1.encode(&#039;ascii&#039;))
    self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)
    self.assertIn(self.no_lifecycle_config, stdout)

  def test_valid_lifecycle(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(contents=self.lifecycle_doc.encode(&#039;ascii&#039;))
    self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_json_obj)

  def test_valid_lifecycle_bucket_style(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(
        contents=self.lifecycle_doc_bucket_style.encode(&#039;ascii&#039;))
    self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_json_obj)

  def test_created_before_lifecycle(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(
        contents=self.lifecycle_created_before_doc.encode(&#039;ascii&#039;))
    self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_created_before_json_obj)

  def test_bad_lifecycle(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(contents=self.bad_doc.encode(&#039;ascii&#039;))
    stderr = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)],
        expected_status=1,
        return_stderr=True)
    self.assertNotIn(&#039;XML lifecycle data provided&#039;, stderr)

  def test_bad_xml_lifecycle(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(contents=self.xml_doc.encode(&#039;ascii&#039;))
    stderr = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)],
        expected_status=1,
        return_stderr=True)
    self.assertIn(&#039;XML lifecycle data provided&#039;, stderr)

  def test_translation_for_falsy_values_works_correctly(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(
        contents=self.lifecycle_with_falsy_condition_values.encode(&#039;ascii&#039;))

    self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)

    # The lifecycle policy we fetch should include all the False- and 0-valued
    # attributes that we just set.
    self.assertRegex(stdout, r&#039;&quot;age&quot;:\s+0&#039;)
    self.assertRegex(stdout, r&#039;&quot;isLive&quot;:\s+false&#039;)
    self.assertRegex(stdout, r&#039;&quot;numNewerVersions&quot;:\s+0&#039;)

  def test_set_lifecycle_and_reset(self):
    &quot;&quot;&quot;Tests setting and turning off lifecycle configuration.&quot;&quot;&quot;
    bucket_uri = self.CreateBucket()
    tmpdir = self.CreateTempDir()
    fpath = self.CreateTempFile(tmpdir=tmpdir,
                                contents=self.lifecycle_doc.encode(&#039;ascii&#039;))
    self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_json_obj)

    fpath = self.CreateTempFile(tmpdir=tmpdir,
                                contents=self.empty_doc1.encode(&#039;ascii&#039;))
    self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, suri(bucket_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket_uri)], return_stdout=True)
    self.assertIn(self.no_lifecycle_config, stdout)

  def test_set_lifecycle_multi_buckets(self):
    &quot;&quot;&quot;Tests setting lifecycle configuration on multiple buckets.&quot;&quot;&quot;
    bucket1_uri = self.CreateBucket()
    bucket2_uri = self.CreateBucket()
    fpath = self.CreateTempFile(contents=self.lifecycle_doc.encode(&#039;ascii&#039;))
    self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;set&#039;, fpath,
         suri(bucket1_uri),
         suri(bucket2_uri)])
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket1_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_json_obj)
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket2_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_json_obj)

  def test_set_lifecycle_wildcard(self):
    &quot;&quot;&quot;Tests setting lifecycle with a wildcarded bucket URI.&quot;&quot;&quot;
    if self.test_api == ApiSelector.XML:
      # This test lists buckets with wildcards, but it is possible that another
      # test being run in parallel (in the same project) deletes a bucket after
      # it is listed in this test. This causes the subsequent XML metadata get
      # for the lifecycle configuration to fail on that just-deleted bucket,
      # even though that bucket is not used directly in this test.
      return unittest.skip(&#039;XML wildcard behavior can cause test to flake &#039;
                           &#039;if a bucket in the same project is deleted &#039;
                           &#039;during execution.&#039;)

    random_prefix = self.MakeRandomTestString()
    bucket1_name = self.MakeTempName(&#039;bucket&#039;, prefix=random_prefix)
    bucket2_name = self.MakeTempName(&#039;bucket&#039;, prefix=random_prefix)
    bucket1_uri = self.CreateBucket(bucket_name=bucket1_name)
    bucket2_uri = self.CreateBucket(bucket_name=bucket2_name)
    # This just double checks that the common prefix of the two buckets is what
    # we think it should be (based on implementation detail of CreateBucket).
    # We want to be careful when setting a wildcard on buckets to make sure we
    # don&#039;t step outside the test buckets to affect other buckets.
    common_prefix = posixpath.commonprefix(
        [suri(bucket1_uri), suri(bucket2_uri)])
    self.assertTrue(
        common_prefix.startswith(
            &#039;gs://%sgsutil-test-test-set-lifecycle-wildcard-&#039; % random_prefix))
    wildcard = &#039;%s*&#039; % common_prefix

    fpath = self.CreateTempFile(contents=self.lifecycle_doc.encode(&#039;ascii&#039;))

    actual_lines = set()
    # Use @Retry as hedge against bucket listing eventual consistency.
    @Retry(AssertionError, tries=3, timeout_secs=1)
    def _Check1():
      stderr = self.RunGsUtil([&#039;lifecycle&#039;, &#039;set&#039;, fpath, wildcard],
                              return_stderr=True)
      actual_lines.update(stderr.splitlines())
      if self._use_gcloud_storage:
        # Gcloud may have unrelated characters on status line
        # as a result of output logic so can&#039;t do direct line string comparison.
        self.assertIn(&#039;Updating %s/...&#039; % suri(bucket1_uri), stderr)
        self.assertIn(&#039;Updating %s/...&#039; % suri(bucket2_uri), stderr)
        status_message = &#039;Updating&#039;
      else:
        expected_lines = set([
            &#039;Setting lifecycle configuration on %s/...&#039; % suri(bucket1_uri),
            &#039;Setting lifecycle configuration on %s/...&#039; % suri(bucket2_uri)
        ])
        self.assertEqual(expected_lines, actual_lines)
        status_message = &#039;Setting lifecycle configuration&#039;
      self.assertEqual(stderr.count(status_message), 2)

    _Check1()

    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket1_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_json_obj)
    stdout = self.RunGsUtil(
        [&#039;lifecycle&#039;, &#039;get&#039;, suri(bucket2_uri)], return_stdout=True)
    self.assertEqual(json.loads(stdout), self.lifecycle_json_obj)


class TestLifecycleUnitTestsWithShim(testcase.ShimUnitTestBase):
  &quot;&quot;&quot;Unit tests for gsutil lifecycle with shim.&quot;&quot;&quot;

  def test_shim_translates_lifecycle_get_correctly(self):
    bucket_uri = self.CreateBucket()
    with SetBotoConfigForTest([(&#039;GSUtil&#039;, &#039;use_gcloud_storage&#039;, &#039;True&#039;),
                               (&#039;GSUtil&#039;, &#039;hidden_shim_mode&#039;, &#039;dry_run&#039;)]):
      with SetEnvironmentForTest({
          &#039;CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL&#039;: &#039;True&#039;,
          &#039;CLOUDSDK_ROOT_DIR&#039;: &#039;fake_dir&#039;,
      }):
        mock_log_handler = self.RunCommand(&#039;lifecycle&#039;,
                                           args=[
                                               &#039;get&#039;,
                                               suri(bucket_uri),
                                           ],
                                           return_log_handler=True)
        info_lines = &#039;\n&#039;.join(mock_log_handler.messages[&#039;info&#039;])
        self.assertIn(
            (&#039;Gcloud Storage Command: {} storage buckets&#039;
             &#039; describe --format=gsutiljson[key=lifecycle_config,empty=\&#039; has&#039;
             &#039; no lifecycle configuration.\&#039;,empty_prefix_key=storage_url]&#039;
             &#039; {}&#039;).format(shim_util._get_gcloud_binary_path(&#039;fake_dir&#039;),
                                 suri(bucket_uri)), info_lines)

  @mock.patch(&#039;gslib.commands.lifecycle.LifecycleCommand._SetLifecycleConfig&#039;,
              new=mock.Mock())
  def test_shim_translates_lifecycle_set_correctly(self):
    with SetBotoConfigForTest([(&#039;GSUtil&#039;, &#039;use_gcloud_storage&#039;, &#039;True&#039;),
                               (&#039;GSUtil&#039;, &#039;hidden_shim_mode&#039;, &#039;dry_run&#039;)]):
      with SetEnvironmentForTest({
          &#039;CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL&#039;: &#039;True&#039;,
          &#039;CLOUDSDK_ROOT_DIR&#039;: &#039;fake_dir&#039;,
      }):
        mock_log_handler = self.RunCommand(&#039;lifecycle&#039;,
                                           args=[
                                               &#039;set&#039;,
                                               &#039;fake-lifecycle-config.json&#039;,
                                               &#039;gs://fake-bucket1&#039;,
                                               &#039;gs://fake-bucket2&#039;,
                                           ],
                                           return_log_handler=True)
        info_lines = &#039;\n&#039;.join(mock_log_handler.messages[&#039;info&#039;])
        self.assertIn((&#039;Gcloud Storage Command: {} storage buckets&#039;
                       &#039; update --lifecycle-file=fake-lifecycle-config.json&#039;
                       &#039; gs://fake-bucket1 gs://fake-bucket2&#039;).format(
                           shim_util._get_gcloud_binary_path(&#039;fake_dir&#039;)),
                      info_lines)
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fopt%2Fgsutil%2Fgslib%2Ftests'>🔙 Back</a></p>