<commit_before># Copyright 2016 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# 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 "AS IS" 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.

"""
Integration test to see if the image promotion process is working for the
Spinnaker Kubernetes integration.

See testable_service/integration_test.py and spinnaker_testing/spinnaker.py
for more details.

The test will use ssh to peek at the spinnaker configuration
to determine the managed project it should verify, and to determine
the spinnaker account name to use when sending it commands.

Sample Usage:
    Assuming you have created $PASSPHRASE_FILE (which you should chmod 400)
    and $CITEST_ROOT points to the root directory of this repository
    (which is . if you execute this from the root)

  PYTHONPATH=$CITEST_ROOT:$CITEST_ROOT/spinnaker \
    python $CITEST_ROOT/spinnaker/spinnaker_system/kube_smoke_test.py \
    --gce_ssh_passphrase_file=$PASSPHRASE_FILE \
    --gce_project=$PROJECT \
    --gce_zone=$ZONE \
    --gce_instance=$INSTANCE
or
  PYTHONPATH=$CITEST_ROOT:$CITEST_ROOT/spinnaker \
    python $CITEST_ROOT/spinnaker/spinnaker_system/kube_smoke_test.py \
    --native_hostname=host-running-smoke-test
"""

# Standard python modules.
import sys

# citest modules.
import citest.kube_testing as kube
import citest.json_contract as jc
import citest.json_predicate as jp
import citest.service_testing as st

# Spinnaker modules.
import spinnaker_testing as sk
import spinnaker_testing.gate as gate
import spinnaker_testing.frigga as frigga
import citest.base


class KubeSmokeTestScenario(sk.SpinnakerTestScenario):
  """Defines the scenario for the smoke test.

  This scenario defines the different test operations.
  We're going to:
    Create a Spinnaker Application
    Create a Spinnaker Load Balancer
    Create a Spinnaker Server Group
    Create a Pipeline with the following stages
      - Find Image
      - Deploy
    Delete each of the above (in reverse order)
  """

  @classmethod
  def new_agent(cls, bindings):
    """Implements citest.service_testing.AgentTestScenario.new_agent."""
    agent = gate.new_agent(bindings)
    agent.default_max_wait_secs = 180
    return agent

  def __init__(self, bindings, agent=None):
    """Constructor.

    Args:
      bindings: [dict] The data bindings to use to configure the scenario.
      agent: [GateAgent] The agent for invoking the test operations on Gate.
    """
    super(KubeSmokeTestScenario, self).__init__(bindings, agent)
    bindings = self.bindings

    # No detail because name length is restricted too much to afford one!
    self.__lb_detail = ''
    self.__lb_name = frigga.Naming.cluster(
        app=bindings['TEST_APP'],
        stack=bindings['TEST_STACK'])

    # We'll call out the app name because it is widely used
    # because it scopes the context of our activities.
    # pylint: disable=invalid-name
    self.TEST_APP = bindings['TEST_APP']

    self.pipeline_id = None

    # We will deploy two images. One with a tag that we want to find,
    # and another that we don't want to find.
    self.__image_registry = 'gcr.io'
    self.__image_repository = 'kubernetes-spinnaker/test-image'
    self.__desired_image_tag = 'validated'
    self.__undesired_image_tag = 'broken'
    self.__desired_image_pattern = '.*{0}'.format(self.__desired_image_tag)

    image_id_format_string = '{0}/{1}:{2}'

    self.__desired_image_id = image_id_format_string.format(
        self.__image_registry,
        self.__image_repository,
        self.__desired_image_tag)

    self.__undesired_image_id = image_id_format_string.format(
        self.__image_registry,
        self.__image_repository,
        self.__undesired_image_tag)

  def create_app(self):
    """Creates OperationContract that creates a new Spinnaker Application."""
    contract = jc.Contract()
    return st.OperationContract(
        self.agent.make_create_app_operation(
            bindings=self.bindings, application=self.TEST_APP,
            account_name=self.bindings['SPINNAKER_KUBE_ACCOUNT']),
        contract=contract)

  def delete_app(self):
    """Creates OperationContract that deletes a new Spinnaker Application."""
    contract = jc.Contract()
    return st.OperationContract(
        self.agent.make_delete_app_operation(
            application=self.TEST_APP,
            account_name=self.bindings['SPINNAKER_KUBE_ACCOUNT']),
        contract=contract)

  def upsert_load_balancer(self):
    """Creates OperationContract for upsertLoadBalancer.

    Calls Spinnaker's upsertLoadBalancer with a configuration, then verifies
    that the expected resources and configurations are visible on Kubernetes.
    See the contract builder for more info on what the expectations are.
    """
    bindings = self.bindings
    payload = self.agent.make_json_payload_from_kwargs(
        job=[{
            'cloudProvider': 'kubernetes',
            'availabilityZones': {'default': ['default']},
            'provider': 'kubernetes',
            'stack': bindings['TEST_STACK'],
            'detail': self.__lb_detail,<commit_msg>fixed kubernetes test account reference (#1242)<commit_after># Copyright 2016 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# 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 "AS IS" 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.

"""
Integration test to see if the image promotion process is working for the
Spinnaker Kubernetes integration.

See testable_service/integration_test.py and spinnaker_testing/spinnaker.py
for more details.

The test will use ssh to peek at the spinnaker configuration
to determine the managed project it should verify, and to determine
the spinnaker account name to use when sending it commands.

Sample Usage:
    Assuming you have created $PASSPHRASE_FILE (which you should chmod 400)
    and $CITEST_ROOT points to the root directory of this repository
    (which is . if you execute this from the root)

  PYTHONPATH=$CITEST_ROOT:$CITEST_ROOT/spinnaker \
    python $CITEST_ROOT/spinnaker/spinnaker_system/kube_smoke_test.py \
    --gce_ssh_passphrase_file=$PASSPHRASE_FILE \
    --gce_project=$PROJECT \
    --gce_zone=$ZONE \
    --gce_instance=$INSTANCE
or
  PYTHONPATH=$CITEST_ROOT:$CITEST_ROOT/spinnaker \
    python $CITEST_ROOT/spinnaker/spinnaker_system/kube_smoke_test.py \
    --native_hostname=host-running-smoke-test
"""

# Standard python modules.
import sys

# citest modules.
import citest.kube_testing as kube
import citest.json_contract as jc
import citest.json_predicate as jp
import citest.service_testing as st

# Spinnaker modules.
import spinnaker_testing as sk
import spinnaker_testing.gate as gate
import spinnaker_testing.frigga as frigga
import citest.base


class KubeSmokeTestScenario(sk.SpinnakerTestScenario):
  """Defines the scenario for the smoke test.

  This scenario defines the different test operations.
  We're going to:
    Create a Spinnaker Application
    Create a Spinnaker Load Balancer
    Create a Spinnaker Server Group
    Create a Pipeline with the following stages
      - Find Image
      - Deploy
    Delete each of the above (in reverse order)
  """

  @classmethod
  def new_agent(cls, bindings):
    """Implements citest.service_testing.AgentTestScenario.new_agent."""
    agent = gate.new_agent(bindings)
    agent.default_max_wait_secs = 180
    return agent

  def __init__(self, bindings, agent=None):
    """Constructor.

    Args:
      bindings: [dict] The data bindings to use to configure the scenario.
      agent: [GateAgent] The agent for invoking the test operations on Gate.
    """
    super(KubeSmokeTestScenario, self).__init__(bindings, agent)
    bindings = self.bindings

    # No detail because name length is restricted too much to afford one!
    self.__lb_detail = ''
    self.__lb_name = frigga.Naming.cluster(
        app=bindings['TEST_APP'],
        stack=bindings['TEST_STACK'])

    # We'll call out the app name because it is widely used
    # because it scopes the context of our activities.
    # pylint: disable=invalid-name
    self.TEST_APP = bindings['TEST_APP']

    self.pipeline_id = None

    # We will deploy two images. One with a tag that we want to find,
    # and another that we don't want to find.
    self.__image_registry = 'gcr.io'
    self.__image_repository = 'kubernetes-spinnaker/test-image'
    self.__desired_image_tag = 'validated'
    self.__undesired_image_tag = 'broken'
    self.__desired_image_pattern = '.*{0}'.format(self.__desired_image_tag)

    image_id_format_string = '{0}/{1}:{2}'

    self.__desired_image_id = image_id_format_string.format(
        self.__image_registry,
        self.__image_repository,
        self.__desired_image_tag)

    self.__undesired_image_id = image_id_format_string.format(
        self.__image_registry,
        self.__image_repository,
        self.__undesired_image_tag)

  def create_app(self):
    """Creates OperationContract that creates a new Spinnaker Application."""
    contract = jc.Contract()
    return st.OperationContract(
        self.agent.make_create_app_operation(
            bindings=self.bindings, application=self.TEST_APP,
            account_name=self.bindings['SPINNAKER_KUBERNETES_ACCOUNT']),
        contract=contract)

  def delete_app(self):
    """Creates OperationContract that deletes a new Spinnaker Application."""
    contract = jc.Contract()
    return st.OperationContract(
        self.agent.make_delete_app_operation(
            application=self.TEST_APP,
            account_name=self.bindings['SPINNAKER_KUBERNETES_ACCOUNT']),
        contract=contract)

  def upsert_load_balancer(self):
    """Creates OperationContract for upsertLoadBalancer.

    Calls Spinnaker's upsertLoadBalancer with a configuration, then verifies
    that the expected resources and configurations are visible on Kubernetes.
    See the contract builder for more info on what the expectations are.
    """
    bindings = self.bindings
    payload = self.agent.make_json_payload_from_kwargs(
        job=[{
            'cloudProvider': 'kubernetes',
            'availabilityZones': {'default': ['default']},
            'provider': 'kubernetes',
            'stack': bindings['TEST_STACK'],
            'detail': self.__lb_detail,
{'fraction_non_alphanumeric': 0.07812218417732925, 'fraction_numerical': 0.003604253018561903, 'mean_word_length': 3.647822445561139, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1069505', 'n_tokens_mistral': 3321, 'n_tokens_neox': 3200, 'n_words': 992}