categ_ansible_test_code.txt

===========Repository Name===========
ansible-role-container-registry
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-container-registry\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 1.6
envlist = docs, linters
skipdist = True

[testenv]
usedevelop = True
install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
setenv = VIRTUAL_ENV={envdir}
deps = -r{toxinidir}/test-requirements.txt
whitelist_externals = bash

[testenv:bindep]
basepython = python3
# Do not install any requirements. We want this to be fast and work even if
# system dependencies are missing, since it's used to tell you what system
# dependencies are missing! This also means that bindep must be installed
# separately, outside of the requirements files.
deps = bindep
commands = bindep test

[testenv:pep8]
basepython = python3
commands =
    # Run hacking/flake8 check for all python files
    bash -c "git ls-files | grep -v releasenotes |  xargs grep --binary-files=without-match \
        --files-with-match '^.!.*python$' \
        --exclude-dir .tox \
        --exclude-dir .git \
        --exclude-dir .eggs \
        --exclude-dir *.egg-info \
        --exclude-dir dist \
        --exclude-dir *lib/python* \
        --exclude-dir doc \
        | xargs flake8 --verbose"

[testenv:ansible-lint]
basepython=python3
commands =
  bash ci-scripts/ansible-lint.sh

[testenv:linters]
basepython = python3
deps =
    -r{toxinidir}/test-requirements.txt
    -r{toxinidir}/ansible-requirements.txt
commands =
    {[testenv:pep8]commands}
    {[testenv:ansible-lint]commands}

[testenv:releasenotes]
basepython = python3
whitelist_externals = bash
commands = bash -c ci-scripts/releasenotes_tox.sh

[testenv:venv]
basepython = python3
commands = {posargs}

[flake8]
# E123, E125 skipped as they are invalid PEP-8.
# E265 deals withs paces inside of comments
show-source = True
ignore = E123,E125,E265
builtins = _




===========Repository Name===========
ansible-role-container-registry
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-container-registry\tests\inventory
===========File Type===========

===========File Content===========
localhost





===========Repository Name===========
ansible-role-container-registry
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-container-registry\tests\test.yml
===========File Type===========
.yml
===========File Content===========
- hosts: localhost
  become: true
  roles:
    - container-registry




===========Repository Name===========
ansible-role-python_venv_build
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-python_venv_build\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
skipsdist = True
envlist = docs,linters,functional


[testenv]
usedevelop = True
install_command =
    pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
commands =
    /usr/bin/find . -type f -name "*.pyc" -delete
passenv =
    COMMON_TESTS_PATH
    HOME
    http_proxy
    HTTP_PROXY
    https_proxy
    HTTPS_PROXY
    no_proxy
    NO_PROXY
    TESTING_BRANCH
    TESTING_HOME
    USER
whitelist_externals =
    bash
setenv =
    PYTHONUNBUFFERED=1
    ROLE_NAME=python_venv_build
    VIRTUAL_ENV={envdir}
    WORKING_DIR={toxinidir}
    # TODO(odyssey4me): remove after debugging is completed
    ANSIBLE_PARAMETERS=-v


[testenv:docs]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands=
    bash -c "rm -rf doc/build"
    doc8 doc
    sphinx-build -b html doc/source doc/build/html


[doc8]
# Settings for doc8:
extensions = .rst


[testenv:releasenotes]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands =
    sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html


# environment used by the -infra templated docs job
[testenv:venv]
basepython = python3
commands =
    {posargs}


[testenv:pep8]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-pep8.sh"


[flake8]
# Ignores the following rules due to how ansible modules work in general
#     F403 'from ansible.module_utils.basic import *' used;
#          unable to detect undefined names
ignore=F403


[testenv:bashate]
commands =
    bash -c "{toxinidir}/tests/common/test-bashate.sh"


# The deps URL should be set to the appropriate git URL.
# In the tests repo itself, the variable is uniquely set to
# the toxinidir so that the role is able to test itself, but
# the tox config is exactly the same as other repositories.
#
# The value for other repositories must be:
# http://git.openstack.org/cgit/openstack/openstack-ansible-tests/plain/test-ansible-deps.txt
# or for a stable branch:
# http://git.openstack.org/cgit/openstack/openstack-ansible-tests/plain/test-ansible-deps.txt?h=stable/queens


[testenv:ansible-syntax]
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-syntax.sh"


[testenv:ansible-lint]
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-lint.sh"


[testenv:functional]
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-functional.sh"


[testenv:linters]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-env-prep.sh"
    {[testenv:pep8]commands}
    {[testenv:bashate]commands}
    {[testenv:ansible-lint]commands}
    {[testenv:ansible-syntax]commands}




===========Repository Name===========
ansible-role-python_venv_build
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-python_venv_build\tests\ansible-role-requirements.yml
===========File Type===========
.yml
===========File Content===========
- name: apt_package_pinning
  src: https://git.openstack.org/openstack/openstack-ansible-apt_package_pinning
  scm: git
  version: master
- name: pip_install
  src: https://git.openstack.org/openstack/openstack-ansible-pip_install
  scm: git
  version: master
- name: openstack_hosts
  src: https://git.openstack.org/openstack/openstack-ansible-openstack_hosts
  scm: git
  version: master
- name: lxc_hosts
  src: https://git.openstack.org/openstack/openstack-ansible-lxc_hosts
  scm: git
  version: master
- name: lxc_container_create
  src: https://git.openstack.org/openstack/openstack-ansible-lxc_container_create
  scm: git
  version: master




===========Repository Name===========
ansible-role-python_venv_build
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-python_venv_build\tests\inventory
===========File Type===========

===========File Content===========
[all]
localhost
container1
container2
container3

[all_containers]
container1
container2
container3




===========Repository Name===========
ansible-role-python_venv_build
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-python_venv_build\tests\test.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2018, Rackspace US, Inc.
#
# 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.

- name: Prepare the host/containers
  import_playbook: common/test-setup-host.yml

- name: Prepare web server on localhost to serve python packages
  hosts: localhost
  connection: local
  become: yes
  any_errors_fatal: yes
  tasks:
    - name: Set venv_build_archive_path and venv_install_source_path
      set_fact:
        venv_build_host_wheel_path: >-
          {%- if ansible_distribution == "Ubuntu" %}
          {%-   set _path = "/var/www/html" %}
          {%- elif ansible_distribution == "CentOS" %}
          {%-   set _path = "/usr/share/nginx/html" %}
          {%- else %}
          {%-   set _path = "/srv/www/htdocs" %}
          {%- endif %}
          {{- _path }}

    - name: Install EPEL gpg keys
      rpm_key:
        key: "http://dl.fedoraproject.org/pub/epel/RPM-GPG-KEY-EPEL-7"
        state: present
      when:
        - ansible_pkg_mgr in ['yum', 'dnf']
      register: _add_yum_keys
      until: _add_yum_keys  is success
      retries: 5
      delay: 2

    - name: Install the EPEL repository
      yum_repository:
        name: epel-nginx
        baseurl: "{{ (centos_epel_mirror | default ('http://download.fedoraproject.org/pub/epel')) ~ '/' ~ ansible_distribution_major_version ~ '/' ~ ansible_architecture }}"
        description: 'Extra Packages for Enterprise Linux 7 - $basearch'
        gpgcheck: yes
        enabled: yes
        state: present
        includepkgs: 'nginx*'
      when:
        - ansible_pkg_mgr in ['yum', 'dnf']
      register: install_epel_repo
      until: install_epel_repo  is success
      retries: 5
      delay: 2

    - name: Install distro packages
      package:
        name: "nginx"
        update_cache: "{{ (ansible_pkg_mgr in ['apt', 'zypper']) | ternary('yes', omit) }}"
      register: install
      until: install  is success
      retries: 5
      delay: 2

    - name: Enable and start nginx
      service:
        name: nginx
        enabled: yes
        daemon_reload: yes
        state: restarted

- name: Verify not using a build host
  hosts: "container1"
  remote_user: root
  any_errors_fatal: yes
  vars:
    venv_pip_packages:
      - "Jinja2==2.10"
    venv_install_destination_path: "/openstack/venvs/test-venv"
  tasks:

    - name: Execute venv install
      include_role:
        name: "python_venv_build"
        private: yes
      vars:
        venv_facts_when_changed:
          - section: "{{ inventory_hostname }}"
            option: "test"
            value: True

    - name: refresh local facts
      setup:
        filter: ansible_local
        gather_subset: "!all"

    - name: Show the ansible_local facts
      debug:
        var: ansible_local

    - name: Verify that the facts were set
      assert:
        that:
          - ansible_local['openstack_ansible'][inventory_hostname]['test'] | bool

    - name: Find files/folders on targets
      find:
        file_type: directory
        get_checksum: no
        recurse: no
        paths:
          - "{{ venv_install_destination_path | dirname }}"
      register: _target_folders

    - name: Compile the folder list from the targets
      set_fact:
        _target_folder_list: "{{ _target_folders['files'] | map(attribute='path') | list }}"

    - name: Show the files/folder from the targets
      debug:
        var: _target_folder_list

    - name: Verify the folder list from the targets
      assert:
        that:
          - "{{ venv_install_destination_path in _target_folder_list }}"

- name: Verify using a build host
  hosts: "container2:container3"
  remote_user: root
  any_errors_fatal: yes
  vars:
    venv_default_pip_packages:
      - "elasticsearch>=6.0.0,<7.0.0"
    venv_pip_packages:
      - "Jinja2==2.10"
    venv_install_destination_path: "/openstack/venvs/test-venv"
    venv_pip_install_args: >-
      --find-links http://{{ hostvars['localhost'].ansible_default_ipv4.address }}
      --trusted-host {{ hostvars['localhost'].ansible_default_ipv4.address }}
    venv_build_host: localhost
    venv_build_host_wheel_path: "{{ hostvars['localhost']['venv_build_host_wheel_path'] }}"
  tasks:

    - name: Execute venv install
      include_role:
        name: "python_venv_build"
        private: yes
      vars:
        venv_facts_when_changed:
          - section: "{{ inventory_hostname }}"
            option: "test"
            value: True

    - name: refresh local facts
      setup:
        filter: ansible_local
        gather_subset: "!all"

    - name: Show the ansible_local facts
      debug:
        var: ansible_local

    - name: Verify that the facts were set
      assert:
        that:
          - ansible_local['openstack_ansible'][inventory_hostname]['test'] | bool

    - name: Find files/folders on targets
      find:
        file_type: directory
        get_checksum: no
        recurse: no
        paths:
          - "{{ venv_install_destination_path | dirname }}"
      register: _target_folders

    - name: Compile the folder list from the targets
      set_fact:
        _target_folder_list: "{{ _target_folders['files'] | map(attribute='path') | list }}"

    - name: Show the files/folder from the targets
      debug:
        var: _target_folder_list

    - name: Verify the folder list from the targets
      assert:
        that:
          - "{{ venv_install_destination_path in _target_folder_list }}"


-----------> External dependency ... import_playbook: common/test-setup-host.yml ... dependent playbook is not in 'test' folder but in 'common' folder 
-----------> Follows a 'prepare' and 'verify' principle 

===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
skipsdist = True
envlist = docs,linters,functional


[testenv]
usedevelop = True
install_command =
    pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
commands =
    /usr/bin/find . -type f -name "*.pyc" -delete
passenv =
    COMMON_TESTS_PATH
    HOME
    http_proxy
    HTTP_PROXY
    https_proxy
    HTTPS_PROXY
    no_proxy
    NO_PROXY
    TESTING_BRANCH
    TESTING_HOME
    USER
whitelist_externals =
    bash
setenv =
    PYTHONUNBUFFERED=1
    ROLE_NAME=systemd_mount
    TEST_IDEMPOTENCE=false
    VIRTUAL_ENV={envdir}
    WORKING_DIR={toxinidir}


[testenv:docs]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands=
    bash -c "rm -rf doc/build"
    doc8 doc
    sphinx-build -b html doc/source doc/build/html


[doc8]
# Settings for doc8:
extensions = .rst


[testenv:releasenotes]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands =
    sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html


# environment used by the -infra templated docs job
[testenv:venv]
basepython = python3
commands =
    {posargs}


[testenv:pep8]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-pep8.sh"


[flake8]
# Ignores the following rules due to how ansible modules work in general
#     F403 'from ansible.module_utils.basic import *' used;
#          unable to detect undefined names
ignore=F403


[testenv:bashate]
commands =
    bash -c "{toxinidir}/tests/common/test-bashate.sh"


[testenv:ansible-syntax]
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-syntax.sh"


[testenv:ansible-lint]
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-lint.sh"


[testenv:functional]
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-functional.sh"


[testenv:linters]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-env-prep.sh"
    {[testenv:pep8]commands}
    {[testenv:bashate]commands}
    {[testenv:ansible-lint]commands}
    {[testenv:ansible-syntax]commands}
    {[testenv:docs]commands}




===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tests\ansible-role-requirements.yaml
===========File Type===========
.yaml
===========File Content===========




===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tests\inventory
===========File Type===========

===========File Content===========
[all]
localhost




===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tests\systemd_init-overrides.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2018, Rackspace US, Inc.
#
# 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.

tempest_run: yes
tempest_venv_tag: "{{ tempest_git_install_branch }}"
tempest_venv_bin: "/opt/tempest_{{ tempest_venv_tag }}/bin"
tempest_log_dir: "/var/log/"
tempest_test_whitelist:
  - tempest.scenario.test_server_basic_ops.TestServerBasicOps.test_server_basic_ops

neutron_provider_networks:
  network_types: "vxlan,flat"
  network_mappings: "flat:eth12"
  network_vxlan_ranges: "1:1000"




===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tests\test-create-btrfs-dev.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2017, BBC R&D
#
# 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.

- name: Configure BTRFS sparse file
  hosts: localhost
  user: root
  become: true
  connection: local
  tasks:
    - name: Install BTRFS packages
      package:
        name: "{{ btrfs_package[ansible_pkg_mgr | lower] }}"
        state: present

    - name: Create base directories
      file:
        path: "/var/lib"
        state: "directory"

    - name: Create sparse file
      command: "truncate -s 1024G /var/lib/sparse-file.img"
      args:
        creates: /var/lib/sparse-file.img
      register: sparse_file

    - name: Format the sparse file
      filesystem:
        fstype: btrfs
        dev: /var/lib/sparse-file.img
      when:
        - sparse_file  is changed
  vars:
    btrfs_package:
      apt: "btrfs-tools"
      yum: "btrfs-progs"
      zypper: "btrfsprogs"

-----------> Does not follow a 'prepare' and 'verify' principle 




===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tests\test-create-nfs-dev.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2017, BBC R&D
#
# 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.

- name: Create an NFS backing store
  hosts: localhost
  user: root
  become: true
  connection: local
  tasks:
    - name: Install NFS packages
      package:
        name: "{{ nfs_package[ansible_distribution.split()[0] | lower] }}"
        state: present

    - name: create the system group for nfs
      group:
        name: "nfs-user"
        gid: "10000"
        state: "present"
        system: "yes"

    - name: Create the system user for nfs
      user:
        name: "nfs-user"
        uid: "10000"
        group: "nfs-user"
        comment: "nfs-user"
        shell: "/bin/false"
        system: "yes"
        createhome: "yes"
        home: "/srv/nfs"

    - name: Create base directories
      file:
        path: "{{ item }}"
        state: "directory"
        owner: "nfs-user"
        group: "nfs-user"
      with_items:
        - "/srv/nfs/test"

    - name: Create exports file
      lineinfile:
        path: /etc/exports
        line: '{{ item }} 127.0.0.1/255.0.0.0(rw,sync,no_subtree_check,insecure,all_squash,anonuid=10000,anongid=10000)'
        owner: root
        group: root
        mode: 0644
        create: yes
      with_items:
        - "/srv/nfs/test"
      register: nfs_exportfs

    - name: Restart nfs-server
      systemd:
        daemon_reload: yes
        name: "nfs-server"
        enabled: "yes"
        state: "restarted"
      when:
        - nfs_exportfs  is changed

    - name: Export NFS
      command: exportfs -rav
      tags:
        - skip_ansible_lint
  vars:
    nfs_package:
      ubuntu: "nfs-kernel-server"
      centos: "nfs-utils"
      opensuse: "nfs-kernel-server"


-----------> Does not follow a 'prepare' and 'verify' principle ... we don't see any assert statements in any of the tasks 

===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tests\test-create-swap-dev.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2017, BBC R&D
#
# 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.

- name: Configure swap sparse file
  hosts: localhost
  user: root
  become: true
  connection: local
  tasks:
    - name: Create swap file
      command: "dd if=/dev/zero of=/var/lib/test-swap.img bs=1M count=128"
      args:
        creates: /var/lib/test-swap.img
      register: create_swap

    - name: Format the swap file
      command: mkswap /var/lib/test-swap.img
      failed_when: false
      when:
        - create_swap  is changed


-----------> Does not follow a 'prepare' and 'verify' principle ... no assert to verify configuration of swap sparse file 

===========Repository Name===========
ansible-role-systemd_mount
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_mount\tests\test.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2018, Rackspace US, Inc.
# Copyright 2018, Logan Vig <logan2211@gmail.com>
#
# 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.

- import_playbook: test-create-swap-dev.yml

- import_playbook: test-create-nfs-dev.yml

- import_playbook: test-create-btrfs-dev.yml

- name: Playbook for role testing
  hosts: localhost
  connection: local
  user: root
  become: true
  roles:
    - role: "systemd_mount"

  post_tasks:
    - name: Ensure mount are mounted
      command: grep -w '{{ item }}' /proc/mounts
      with_items:
        - /var/lib/sparse-file
        - /var/lib/test
      tags:
        - skip_ansible_lint

    - name: Ensure swap is enabled
      shell: swapon | grep -w '/var/lib/test-swap.img'
      tags:
        - skip_ansible_lint

  vars:
    systemd_mounts:
      - what: '/var/lib/sparse-file.img'
        where: '/var/lib/sparse-file'
        type: 'btrfs'
        options: 'loop'
        state: 'started'
        enabled: true
        config_overrides:
          Unit:
            ConditionPathExists: '/var/lib/sparse-file.img'

      - what: "/var/lib/test-swap.img"
        priority: "0"
        options: "%%"
        type: "swap"
        state: 'started'
        enabled: true

      - what: "127.0.0.1:/srv/nfs/test"
        where: "/var/lib/test"
        type: "nfs"
        options: "_netdev,auto"
        state: 'started'
        enabled: true
        config_overrides:
          Unit:
            After:
              ? network.target
              ? network-online.target
            Wants: network-online.target




===========Repository Name===========
ansible-role-systemd_networkd
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_networkd\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
skipsdist = True
envlist = docs,linters,functional


[testenv]
usedevelop = True
install_command =
    pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
commands =
    /usr/bin/find . -type f -name "*.pyc" -delete
passenv =
    COMMON_TESTS_PATH
    HOME
    http_proxy
    HTTP_PROXY
    https_proxy
    HTTPS_PROXY
    no_proxy
    NO_PROXY
    TESTING_BRANCH
    TESTING_HOME
    USER
whitelist_externals =
    bash
setenv =
    PYTHONUNBUFFERED=1
    ROLE_NAME=systemd_networkd
    TEST_IDEMPOTENCE=false
    VIRTUAL_ENV={envdir}
    WORKING_DIR={toxinidir}


[testenv:docs]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands=
    bash -c "rm -rf doc/build"
    doc8 doc
    sphinx-build -b html doc/source doc/build/html


[doc8]
# Settings for doc8:
extensions = .rst


[testenv:releasenotes]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands =
    sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html


# environment used by the -infra templated docs job
[testenv:venv]
basepython = python3
commands =
    {posargs}


[testenv:pep8]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-pep8.sh"


[flake8]
# Ignores the following rules due to how ansible modules work in general
#     F403 'from ansible.module_utils.basic import *' used;
#          unable to detect undefined names
ignore=F403


[testenv:bashate]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-bashate.sh"


[testenv:ansible-syntax]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-syntax.sh"


[testenv:ansible-lint]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-lint.sh"


[testenv:functional]
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-functional.sh"


[testenv:linters]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-env-prep.sh"
    {[testenv:pep8]commands}
    {[testenv:bashate]commands}
    {[testenv:ansible-lint]commands}
    {[testenv:ansible-syntax]commands}
    {[testenv:docs]commands}


-----------> We need to skip .ini files 

===========Repository Name===========
ansible-role-systemd_networkd
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_networkd\tests\ansible-role-requirements.yaml
===========File Type===========
.yaml
===========File Content===========




===========Repository Name===========
ansible-role-systemd_networkd
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_networkd\tests\inventory
===========File Type===========

===========File Content===========
[all]
localhost




===========Repository Name===========
ansible-role-systemd_networkd
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_networkd\tests\systemd_init-overrides.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2018, Rackspace US, Inc.
#
# 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.

tempest_run: yes
tempest_venv_tag: "{{ tempest_git_install_branch }}"
tempest_venv_bin: "/opt/tempest_{{ tempest_venv_tag }}/bin"
tempest_log_dir: "/var/log/"
tempest_test_whitelist:
  - tempest.scenario.test_server_basic_ops.TestServerBasicOps.test_server_basic_ops

neutron_provider_networks:
  network_types: "vxlan,flat"
  network_mappings: "flat:eth12"
  network_vxlan_ranges: "1:1000"


-----------> We need to skip YML files that has no 'name:' tag ... 'name' reflects a playbook executing some steps 


===========Repository Name===========
ansible-role-systemd_networkd
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-systemd_networkd\tests\test.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2018, Rackspace US, Inc.
# Copyright 2018, Logan Vig <logan2211@gmail.com>
#
# 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.

- name: Playbook for role testing
  hosts: localhost
  connection: local
  become: true
  gather_facts: true
  roles:
    - role: "systemd_networkd"
  vars:
    systemd_run_networkd: yes
    systemd_resolved:
      DNS: "208.67.222.222"
      FallbackDNS: "8.8.8.8"
      Cache: yes
    systemd_netdevs:
      - NetDev:
          Name: dummy0
          Kind: dummy
      - NetDev:
          Name: dummy1
          Kind: dummy
      - NetDev:
          Name: bond0
          Kind: bond
        Bond:
          Mode: 802.3ad
          TransmitHashPolicy: layer3+4
          MIIMonitorSec: 1s
          LACPTransmitRate: fast
      - NetDev:
          Name: br-dummy
          Kind: bridge
      - NetDev:
          Name: dummy2
          Kind: dummy
      - NetDev:
          Name: br-test
          Kind: bridge
    systemd_networks:
      - interface: "dummy0"
        bond: "bond0"
        mtu: 9000
      - interface: "dummy1"
        bond: "bond0"
        mtu: 9000
      - interface: "bond0"
        bridge: "br-dummy"
        mtu: 9000
      - interface: "br-dummy"
        address: "10.0.0.100"
        netmask: "255.255.255.0"
        gateway: "10.0.0.1"
        mtu: 9000
        usedns: true
        static_routes:
          - gateway: "10.1.0.1"
            cidr: "10.1.0.0/24"
        config_overrides:
          Network:
            ConfigureWithoutCarrier: true
      - interface: "dummy2"
        bridge: "br-test"
      - interface: "br-test"
        address: "10.1.0.1"
        netmask: "255.255.255.0"


- name: Test networkd
  hosts: localhost
  connection: local
  become: true
  gather_facts: true
  tasks:
    - name: Interface check
      assert:
        that:
          - ansible_dummy0['active'] == true
          - ansible_dummy0['type'] == 'ether'
          - ansible_dummy0['mtu'] == 9000
          - ansible_dummy1['active'] == true
          - ansible_dummy1['type'] == 'ether'
          - ansible_dummy1['mtu'] == 9000
          - ansible_dummy2['active'] == true
          - ansible_dummy2['type'] == 'ether'
    - name: Bond check
      assert:
        that:
          - ansible_bond0['active'] == true
          - ansible_bond0['type'] == 'bonding'
          - ansible_bond0['mtu'] == 9000
    - name: Bridge check
      assert:
        that:
          - ansible_br_dummy['active'] == true
          - ansible_br_dummy['type'] == 'bridge'
          - ansible_br_dummy['ipv4']['address'] == '10.0.0.100'
          - ansible_br_dummy['ipv4']['netmask'] == '255.255.255.0'
    - name: Bridge check
      assert:
        that:
          - ansible_br_test['active'] == true
          - ansible_br_test['type'] == 'bridge'
          - ansible_br_test['ipv4']['address'] == '10.1.0.1'
          - ansible_br_test['ipv4']['netmask'] == '255.255.255.0'


- name: Playbook for role testing with cleanup
  hosts: localhost
  connection: local
  become: true
  gather_facts: true
  roles:
    - role: "systemd_networkd"
  post_tasks:
    - name: Interface check
      assert:
        that:
          - ansible_br_test is defined
          - ansible_dummy2['active'] == true
          - ansible_dummy2['type'] == 'ether'
    - name: Bridge check
      assert:
        that:
          - ansible_br_test['active'] == true
          - ansible_br_test['type'] == 'bridge'
          - ansible_br_test['ipv4']['address'] == '10.1.0.1'
          - ansible_br_test['ipv4']['netmask'] == '255.255.255.0'
  vars:
    systemd_interface_cleanup: true
    systemd_run_networkd: yes
    systemd_netdevs:
      - NetDev:
          Name: dummy2
          Kind: dummy
      - NetDev:
          Name: br-test
          Kind: bridge
    systemd_networks:
      - interface: "dummyX"
        bridge: "br-test"
      - interface: "br-test"
        address: "10.1.0.1"
        netmask: "255.255.255.0"




===========Repository Name===========
ansible-role-tripleo-modify-image
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ansible-role-tripleo-modify-image\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
envlist = docs, linters
skipdist = True

[testenv]
usedevelop = True
install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
setenv = VIRTUAL_ENV={envdir}
deps = -r{toxinidir}/test-requirements.txt
whitelist_externals = bash

[testenv:bindep]
basepython = python3
# Do not install any requirements. We want this to be fast and work even if
# system dependencies are missing, since it's used to tell you what system
# dependencies are missing! This also means that bindep must be installed
# separately, outside of the requirements files.
deps = bindep
commands = bindep test

[testenv:pep8]
basepython = python3
commands =
    # Run hacking/flake8 check for all python files
    bash -c "git ls-files | grep -v releasenotes |  xargs grep --binary-files=without-match \
        --files-with-match '^.!.*python$' \
        --exclude-dir .tox \
        --exclude-dir .git \
        --exclude-dir .eggs \
        --exclude-dir *.egg-info \
        --exclude-dir dist \
        --exclude-dir *lib/python* \
        --exclude-dir doc \
        | xargs flake8 --verbose"

[testenv:ansible-lint]
basepython=python3
commands =
  bash ci-scripts/ansible-lint.sh

[testenv:linters]
basepython = python3
deps =
    -r{toxinidir}/test-requirements.txt
    -r{toxinidir}/ansible-requirements.txt
commands =
    {[testenv:pep8]commands}
    {[testenv:ansible-lint]commands}

[testenv:releasenotes]
basepython = python3
whitelist_externals = bash
commands = bash -c ci-scripts/releasenotes_tox.sh

[testenv:venv]
basepython = python3
commands = {posargs}

[flake8]
# E123, E125 skipped as they are invalid PEP-8.
# E265 deals with spaces inside of comments
show-source = True
ignore = E123,E125,E265
builtins = _




===========Repository Name===========
bifrost
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\bifrost\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
envlist = py35,py27,docs,pep8
skipsdist = True

[testenv]
usedevelop = True
install_command = pip install -U -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
setenv =
   VIRTUAL_ENV={envdir}
   PYTHONWARNINGS=default::DeprecationWarning
deps = -r{toxinidir}/requirements.txt
       -r{toxinidir}/test-requirements.txt
commands = python setup.py test --slowest --testr-args='{posargs}'

[testenv:pep8]
basepython = python3
commands = flake8
           doc8 doc/source releasenotes/source README.rst CONTRIBUTING.rst MISSION.rst HACKING.rst

[testenv:venv]
basepython = python3
deps =
  -r{toxinidir}/test-requirements.txt
  -r{toxinidir}/doc/requirements.txt
commands = {posargs}

[testenv:cover]
basepython = python3
commands = python setup.py test --coverage --testr-args='{posargs}'

[testenv:docs]
basepython = python3
deps =
  -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt}
  -r{toxinidir}/requirements.txt
  -r{toxinidir}/doc/requirements.txt
commands = sphinx-build -W -b html doc/source doc/build/html

[testenv:debug]
basepython = python3
commands = oslo_debug_helper -t bifrost/tests {posargs}

[testenv:releasenotes]
basepython = python3
deps =
  -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt}
  -r{toxinidir}/doc/requirements.txt
commands = sphinx-build -a -E -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html

[testenv:debug-py27]
basepython = python2.7
commands = oslo_debug_helper -t bifrost/tests {posargs}

[testenv:debug-py35]
basepython = python3.5
commands = oslo_debug_helper -t bifrost/tests {posargs}

[flake8]
show-source = True
ignore = F403,H102,H303
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build,os_ironic.py,os_ironic_node.py,os_ironic_inspect.py,os_keystone_service.py

[testenv:lower-constraints]
basepython = python3
deps =
  -c{toxinidir}/lower-constraints.txt
  -r{toxinidir}/test-requirements.txt
  -r{toxinidir}/requirements.txt




===========Repository Name===========
bifrost
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\bifrost\bifrost\tests\base.py
===========File Type===========
.py
===========File Content===========
# -*- coding: utf-8 -*-

# Copyright 2010-2011 OpenStack Foundation
# Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
#
# 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.


import testtools


class TestCase(testtools.TestCase):
    """Test case base class for all unit tests."""

    def setUp(self):
        super(TestCase, self).setUp()




===========Repository Name===========
bifrost
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\bifrost\bifrost\tests\utils.py
===========File Type===========
.py
===========File Content===========
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
#
# 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.

import contextlib
import tempfile

from bifrost import inventory


@contextlib.contextmanager
def temporary_file(file_data):
    file = None
    file = tempfile.NamedTemporaryFile(mode='w')
    file.write(file_data)
    file.flush()

    try:
        yield file.name
    finally:
        if file is not None:
            file.close()


def bifrost_csv_conversion(csv_data):
    # TODO(TheJulia): To call prep feels like a bug and should be fixed.
    (groups, hostvars) = inventory._prepare_inventory()
    with temporary_file(csv_data) as file:
        (groups, hostvars) = inventory._process_baremetal_csv(
            file,
            groups,
            hostvars)
    # NOTE(TheJulia): Returning separately so the file is closed first
    return (groups, hostvars)


def bifrost_data_conversion(data):
    (groups, hostvars) = inventory._prepare_inventory()
    with temporary_file(data) as file:
        (groups, hostvars) = inventory._process_baremetal_data(
            file,
            groups,
            hostvars)
    return (groups, hostvars)




===========Repository Name===========
browbeat
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\browbeat\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
envlist = py27,py35,py36,py37,linters,dist,insights
skipsdist = True

[testenv]
usedevelop = True
install_command = pip install -U {opts} {packages}
setenv =
   VIRTUAL_ENV={envdir}
deps = -r{toxinidir}/test-requirements.txt
commands = python setup.py test

[testenv:linters]
# py3 linters are able to stop more than py2 ones
basepython = python3
whitelist_externals = bash
extras = insights
commands =
  {[testenv:pep8]commands}
  bash -c "cd ansible; find . -type f -regex '.*.y[a]?ml' -print0 | xargs -t -n1 -0 \
    ansible-lint \
    -x ANSIBLE0012,ANSIBLE0006,ANSIBLE0007,ANSIBLE0016,ANSIBLE0019" \
    --exclude=rally
  pykwalify -d browbeat-config.yaml -s browbeat/schema/browbeat.yml
  pykwalify -d browbeat-complete.yaml -s browbeat/schema/browbeat.yml
  bash -c "set -e; for config in $(ls conf/); do \
    echo conf/$config; pykwalify -d conf/$config -s browbeat/schema/browbeat.yml; done"

[testenv:dist]
basepython = python3
# reuse linters environment to lower footprint on dev machines
envdir = {toxworkdir}/linters
# test that we can build a valid package
commands =
  python setup.py sdist bdist_wheel
  python -m twine check dist/*

[testenv:insights]
commands =
  pip check
  pip install .[insights]
  pip check

[testenv:pep8]
basepython = python3
commands = flake8 {posargs}

[testenv:venv]
basepython = python3
commands = {posargs}

[testenv:py27]
basepython = python2.7
commands = pytest {posargs}

[testenv:py35]
basepython = python3.5
commands = pytest {posargs}

[testenv:py36]
basepython = python3.6
commands = pytest {posargs}

[testenv:py37]
basepython = python3.7
commands = pytest {posargs}

[testenv:cover]
commands = python setup.py test --coverage --testr-args={posargs}

[testenv:docs]
basepython = python3
commands = python setup.py build_sphinx

[testenv:debug]
basepython = python3
commands = oslo_debug_helper {posargs}

[flake8]
# E123, E125 skipped as they are invalid PEP-8.
show-source = True
ignore = E123,E125,E226,E302,E41,E231,E203,H233,H306,H238,H236,H404,H405,W504
max-line-length = 100
builtins = _
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build,ansible/*,.browbeat-venv,.perfkit-venv,.rally-venv,.shaker-venv




===========Repository Name===========
browbeat
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\browbeat\tests\test_config.py
===========File Type===========
.py
===========File Content===========
#   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.

import pytest
import yaml

from browbeat.config import load_browbeat_config
from browbeat.config import _validate_yaml

test_browbeat_configs = {
    "tests/data/valid_browbeat.yml": True,
    "tests/data/invalid_browbeat.yml": False,
    "tests/data/invalid_browbeat_workload.yml": False
}


@pytest.mark.parametrize("config", test_browbeat_configs.keys())
def test_load_browbeat_config(config):
    """Tests valid and invalid Browbeat configuration."""
    if test_browbeat_configs[config]:
        # Valid configuration (No exception)
        loaded_config = load_browbeat_config(config)
        assert loaded_config["browbeat"]["cloud_name"] == "browbeat-test"
    else:
        # Invalid configuration, test for exception
        with pytest.raises(Exception) as exception_data:
            load_browbeat_config(config)
        assert "SchemaError" in str(exception_data)


@pytest.mark.parametrize("schema", ["perfkit", "rally", "shaker", "yoda"])
def test__validate_yaml(schema):
    """Tests valid and invalid Browbeat workload configurations."""
    with open("tests/data/workloads.yml", "r") as config_file:
        config_data = yaml.safe_load(config_file)

    for workload_config in config_data[schema]:
        if workload_config["valid"]:
            # Valid configuration (No exception)
            _validate_yaml(schema, workload_config["data"])
        else:
            # Invalid configuration, test for exception
            with pytest.raises(Exception) as exception_data:
                _validate_yaml(schema, workload_config["data"])
            assert "SchemaError" in str(exception_data)

--------------> Dependence on external config data (from browbeat.config import load_browbeat_config)


===========Repository Name===========
fuel-ccp-installer
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\fuel-ccp-installer\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 1.6
skipsdist = True
envlist = bashate, pep8

[testenv]
deps =
    -r{toxinidir}/requirements.txt
    -r{toxinidir}/test-requirements.txt

[testenv:doc8]
commands = doc8 doc

[testenv:docs]
whitelist_externals = /bin/rm
commands =
  /bin/rm -rf doc/build
  python setup.py build_sphinx

[doc8]
# Settings for doc8:
# Ignore target directories
ignore-path = doc/build*
# File extensions to use
extensions = .rst,.txt
# Maximal line length should be 79 but we have some overlong lines.
# Let's not get far more in.
max-line-length = 80
# Disable some doc8 checks:
# D000: Check RST validity (cannot handle lineos directive)
ignore = D000

[testenv:bashate]
whitelist_externals = bash
commands = bash -c "find {toxinidir} -type f -name '*.sh' -not -path '*/.tox/*' -print0 | xargs -0 bashate -v"

[testenv:pep8]
usedevelop = False
whitelist_externals = bash
commands =
    bash -c "find {toxinidir}/* -type f -name '*.py' -print0 | xargs -0 flake8"

[testenv:venv]
commands = {posargs}

[flake8]
show-source = true
builtins = _
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,tools




===========Repository Name===========
ironic-lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 1.8
skipsdist = True
envlist = py3,py27,pep8

[testenv]
install_command = pip install {opts} {packages}
usedevelop = True
setenv = VIRTUAL_ENV={envdir}
         PYTHONDONTWRITEBYTECODE = 1
         LANGUAGE=en_US
         TESTS_DIR=./ironic_lib/tests/
deps =
  -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt}
  -r{toxinidir}/test-requirements.txt
  -r{toxinidir}/requirements.txt
commands = stestr run {posargs}

[flake8]
show-source = True
ignore = E129
exclude = .venv,.tox,dist,doc,*.egg,.update-venv
import-order-style = pep8
application-import-names = ironic_lib
# [H106] Don't put vim configuration in source files.
# [H203] Use assertIs(Not)None to check for None.
# [H204] Use assert(Not)Equal to check for equality.
# [H205] Use assert(Greater|Less)(Equal) for comparison.
# [H210] Require 'autospec', 'spec', or 'spec_set' in mock.patch/mock.patch.object calls
# [H904] Delay string interpolations at logging calls.
enable-extensions=H106,H203,H204,H205,H210,H904

[testenv:pep8]
basepython = python3
commands =
    flake8 {posargs}
    doc8 README.rst doc/source --ignore D001

[testenv:cover]
basepython = python3
setenv = VIRTUALENV={envdir}
         LANGUAGE=en_US
         PYTHON=coverage run --source ironic_lib --omit='*tests*' --parallel-mode
commands =
  coverage erase
  stestr run {posargs}
  coverage combine
  coverage report --omit='*tests*'
  coverage html -d ./cover --omit='*tests*'

[testenv:venv]
basepython = python3
commands = {posargs}

[testenv:docs]
basepython = python3
setenv = PYTHONHASHSEED=0
sitepackages = False
envdir = {toxworkdir}/venv
commands =
  python setup.py build_sphinx

[testenv:lower-constraints]
basepython = python3
deps =
  -c{toxinidir}/lower-constraints.txt
  -r{toxinidir}/test-requirements.txt
  -r{toxinidir}/requirements.txt




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\base.py
===========File Type===========
.py
===========File Content===========
# Copyright 2017 Cisco Systems, Inc
#
# 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.

"""Common utilities and classes across all unit tests."""

import subprocess

from oslo_concurrency import processutils
from oslo_config import fixture as config_fixture
from oslotest import base as test_base

from ironic_lib import utils


class IronicLibTestCase(test_base.BaseTestCase):
    """Test case base class for all unit tests except callers of utils.execute.

    This test class prevents calls to the utils.execute() /
    processutils.execute() and similar functions.
    """

    # By default block execution of utils.execute() and related functions.
    block_execute = True

    def setUp(self):
        super(IronicLibTestCase, self).setUp()

        # Make sure config overrides do not leak for test to test.
        self.cfg_fixture = self.useFixture(config_fixture.Config())

        # Ban running external processes via 'execute' like functions. If the
        # patched function is called, an exception is raised to warn the
        # tester.
        if self.block_execute:
            # NOTE(jlvillal): Intentionally not using mock as if you mock a
            # mock it causes things to not work correctly. As doing an
            # autospec=True causes strangeness. By using a simple function we
            # can then mock it without issue.
            self.patch(processutils, 'execute', do_not_call)
            self.patch(subprocess, 'call', do_not_call)
            self.patch(subprocess, 'check_call', do_not_call)
            self.patch(subprocess, 'check_output', do_not_call)
            self.patch(utils, 'execute', do_not_call)

            # subprocess.Popen is a class
            self.patch(subprocess, 'Popen', DoNotCallPopen)


def do_not_call(*args, **kwargs):
    """Helper function to raise an exception if it is called"""
    raise Exception(
        "Don't call ironic_lib.utils.execute() / "
        "processutils.execute() or similar functions in tests!")


class DoNotCallPopen(object):
    """Helper class to mimic subprocess.popen()

    It's job is to raise an exception if it is called. We create stub functions
    so mocks that use autospec=True will work.
    """
    def __init__(self, *args, **kwargs):
        do_not_call(*args, **kwargs)

    def communicate(self, input=None):
        pass

    def kill(self):
        pass

    def poll(self):
        pass

    def terminate(self):
        pass

    def wait(self):
        pass




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\test_base.py
===========File Type===========
.py
===========File Content===========
# 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.

import subprocess

import mock
from oslo_concurrency import processutils

from ironic_lib.tests import base
from ironic_lib import utils


class BlockExecuteTestCase(base.IronicLibTestCase):
    """Test to ensure we block access to the 'execute' type functions"""

    def test_exception_raised_for_execute(self):
        execute_functions = (processutils.execute, subprocess.Popen,
                             subprocess.call, subprocess.check_call,
                             subprocess.check_output, utils.execute)

        for function_name in execute_functions:
            exc = self.assertRaises(
                Exception,
                function_name,
                ["echo", "%s" % function_name])  # noqa
            # Have to use 'noqa' as we are raising plain Exception and we will
            # get H202 error in 'pep8' check.

            self.assertEqual(
                "Don't call ironic_lib.utils.execute() / "
                "processutils.execute() or similar functions in tests!",
                "%s" % exc)

    @mock.patch.object(utils, "execute", autospec=True)
    def test_can_mock_execute(self, mock_exec):
        # NOTE(jlvillal): We had discovered an issue where mocking wasn't
        # working because we had used a mock to block access to the execute
        # functions. This caused us to "mock a mock" and didn't work correctly.
        # We want to make sure that we can mock our execute functions even with
        # our "block execute" code.
        utils.execute("ls")
        utils.execute("echo")
        self.assertEqual(2, mock_exec.call_count)

    @mock.patch.object(processutils, "execute", autospec=True)
    def test_exception_raised_for_execute_parent_mocked(self, mock_exec):
        # Make sure that even if we mock the parent execute function, that we
        # still get an exception for a child. So in this case
        # ironic_lib.utils.execute() calls processutils.execute(). Make sure an
        # exception is raised even though we mocked processutils.execute()
        exc = self.assertRaises(
            Exception,
            utils.execute,
            "ls")  # noqa
        # Have to use 'noqa' as we are raising plain Exception and we will get
        # H202 error in 'pep8' check.

        self.assertEqual(
            "Don't call ironic_lib.utils.execute() / "
            "processutils.execute() or similar functions in tests!",
            "%s" % exc)


class DontBlockExecuteTestCase(base.IronicLibTestCase):
    """Ensure we can turn off blocking access to 'execute' type functions"""

    # Don't block the execute function
    block_execute = False

    @mock.patch.object(processutils, "execute", autospec=True)
    def test_no_exception_raised_for_execute(self, mock_exec):
        # Make sure we can call ironic_lib.utils.execute() even though we
        # didn't mock it. We do mock processutils.execute() so we don't
        # actually execute anything.
        utils.execute("ls")
        utils.execute("echo")
        self.assertEqual(2, mock_exec.call_count)



-----------> Need to filter out Python files altogether ... Python modules need to be skipped 


===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\test_disk_partitioner.py
===========File Type===========
.py
===========File Content===========
# Copyright 2014 Red Hat, 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.

import eventlet
import mock
from testtools.matchers import HasLength

from ironic_lib import disk_partitioner
from ironic_lib import exception
from ironic_lib.tests import base
from ironic_lib import utils


class DiskPartitionerTestCase(base.IronicLibTestCase):

    def test_add_partition(self):
        dp = disk_partitioner.DiskPartitioner('/dev/fake')
        dp.add_partition(1024)
        dp.add_partition(512, fs_type='linux-swap')
        dp.add_partition(2048, boot_flag='boot')
        dp.add_partition(2048, boot_flag='bios_grub')
        expected = [(1, {'boot_flag': None,
                         'extra_flags': None,
                         'fs_type': '',
                         'type': 'primary',
                         'size': 1024}),
                    (2, {'boot_flag': None,
                         'extra_flags': None,
                         'fs_type': 'linux-swap',
                         'type': 'primary',
                         'size': 512}),
                    (3, {'boot_flag': 'boot',
                         'extra_flags': None,
                         'fs_type': '',
                         'type': 'primary',
                         'size': 2048}),
                    (4, {'boot_flag': 'bios_grub',
                         'extra_flags': None,
                         'fs_type': '',
                         'type': 'primary',
                         'size': 2048})]
        partitions = [(n, p) for n, p in dp.get_partitions()]
        self.assertThat(partitions, HasLength(4))
        self.assertEqual(expected, partitions)

    @mock.patch.object(disk_partitioner.DiskPartitioner, '_exec',
                       autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    def test_commit(self, mock_utils_exc, mock_disk_partitioner_exec):
        dp = disk_partitioner.DiskPartitioner('/dev/fake')
        fake_parts = [(1, {'boot_flag': None,
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1}),
                      (2, {'boot_flag': 'boot',
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1}),
                      (3, {'boot_flag': 'bios_grub',
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1}),
                      (4, {'boot_flag': 'boot',
                           'extra_flags': ['prep', 'fake-flag'],
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1})]
        with mock.patch.object(dp, 'get_partitions', autospec=True) as mock_gp:
            mock_gp.return_value = fake_parts
            mock_utils_exc.return_value = ('', '')
            dp.commit()

        mock_disk_partitioner_exec.assert_called_once_with(
            mock.ANY, 'mklabel', 'msdos',
            'mkpart', 'fake-type', 'fake-fs-type', '1', '2',
            'mkpart', 'fake-type', 'fake-fs-type', '2', '3',
            'set', '2', 'boot', 'on',
            'mkpart', 'fake-type', 'fake-fs-type', '3', '4',
            'set', '3', 'bios_grub', 'on',
            'mkpart', 'fake-type', 'fake-fs-type', '4', '5',
            'set', '4', 'boot', 'on', 'set', '4', 'prep', 'on',
            'set', '4', 'fake-flag', 'on')
        mock_utils_exc.assert_called_once_with(
            'fuser', '/dev/fake', run_as_root=True,
            check_exit_code=[0, 1])

    @mock.patch.object(eventlet.greenthread, 'sleep', lambda seconds: None)
    @mock.patch.object(disk_partitioner.DiskPartitioner, '_exec',
                       autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    def test_commit_with_device_is_busy_once(self, mock_utils_exc,
                                             mock_disk_partitioner_exec):
        dp = disk_partitioner.DiskPartitioner('/dev/fake')
        fake_parts = [(1, {'boot_flag': None,
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1}),
                      (2, {'boot_flag': 'boot',
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1})]
        # Test as if the 'psmisc' version of 'fuser' which has stderr output
        fuser_outputs = iter([(" 10000 10001", '/dev/fake:\n'), ('', '')])

        with mock.patch.object(dp, 'get_partitions', autospec=True) as mock_gp:
            mock_gp.return_value = fake_parts
            mock_utils_exc.side_effect = fuser_outputs
            dp.commit()

        mock_disk_partitioner_exec.assert_called_once_with(
            mock.ANY, 'mklabel', 'msdos',
            'mkpart', 'fake-type', 'fake-fs-type', '1', '2',
            'mkpart', 'fake-type', 'fake-fs-type', '2', '3',
            'set', '2', 'boot', 'on')
        mock_utils_exc.assert_called_with(
            'fuser', '/dev/fake', run_as_root=True,
            check_exit_code=[0, 1])
        self.assertEqual(2, mock_utils_exc.call_count)

    @mock.patch.object(eventlet.greenthread, 'sleep', lambda seconds: None)
    @mock.patch.object(disk_partitioner.DiskPartitioner, '_exec',
                       autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    def test_commit_with_device_is_always_busy(self, mock_utils_exc,
                                               mock_disk_partitioner_exec):
        dp = disk_partitioner.DiskPartitioner('/dev/fake')
        fake_parts = [(1, {'boot_flag': None,
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1}),
                      (2, {'boot_flag': 'boot',
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1})]

        with mock.patch.object(dp, 'get_partitions', autospec=True) as mock_gp:
            mock_gp.return_value = fake_parts
            # Test as if the 'busybox' version of 'fuser' which does not have
            # stderr output
            mock_utils_exc.return_value = ("10000 10001", '')
            self.assertRaises(exception.InstanceDeployFailure, dp.commit)

        mock_disk_partitioner_exec.assert_called_once_with(
            mock.ANY, 'mklabel', 'msdos',
            'mkpart', 'fake-type', 'fake-fs-type', '1', '2',
            'mkpart', 'fake-type', 'fake-fs-type', '2', '3',
            'set', '2', 'boot', 'on')
        mock_utils_exc.assert_called_with(
            'fuser', '/dev/fake', run_as_root=True,
            check_exit_code=[0, 1])
        self.assertEqual(20, mock_utils_exc.call_count)

    # Mock the eventlet.greenthread.sleep for the looping_call
    @mock.patch.object(eventlet.greenthread, 'sleep', lambda seconds: None)
    @mock.patch.object(disk_partitioner.DiskPartitioner, '_exec',
                       autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    def test_commit_with_device_disconnected(self, mock_utils_exc,
                                             mock_disk_partitioner_exec):
        dp = disk_partitioner.DiskPartitioner('/dev/fake')
        fake_parts = [(1, {'boot_flag': None,
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1}),
                      (2, {'boot_flag': 'boot',
                           'extra_flags': None,
                           'fs_type': 'fake-fs-type',
                           'type': 'fake-type',
                           'size': 1})]

        with mock.patch.object(dp, 'get_partitions', autospec=True) as mock_gp:
            mock_gp.return_value = fake_parts
            mock_utils_exc.return_value = ('', "Specified filename /dev/fake"
                                               " does not exist.")
            self.assertRaises(exception.InstanceDeployFailure, dp.commit)

        mock_disk_partitioner_exec.assert_called_once_with(
            mock.ANY, 'mklabel', 'msdos',
            'mkpart', 'fake-type', 'fake-fs-type', '1', '2',
            'mkpart', 'fake-type', 'fake-fs-type', '2', '3',
            'set', '2', 'boot', 'on')
        mock_utils_exc.assert_called_with(
            'fuser', '/dev/fake', run_as_root=True,
            check_exit_code=[0, 1])
        self.assertEqual(20, mock_utils_exc.call_count)




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\test_disk_utils.py
===========File Type===========
.py
===========File Content===========
# Copyright 2014 Red Hat, 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.

import gzip
import os
import shutil
import stat
import tempfile

import mock
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_serialization import base64
from oslo_utils import imageutils
import requests

from ironic_lib import disk_partitioner
from ironic_lib import disk_utils
from ironic_lib import exception
from ironic_lib.tests import base
from ironic_lib import utils

CONF = cfg.CONF


@mock.patch.object(utils, 'execute', autospec=True)
class ListPartitionsTestCase(base.IronicLibTestCase):

    def test_correct(self, execute_mock):
        output = """
BYT;
/dev/sda:500107862016B:scsi:512:4096:msdos:ATA HGST HTS725050A7:;
1:1.00MiB:501MiB:500MiB:ext4::boot;
2:501MiB:476940MiB:476439MiB:::;
"""
        expected = [
            {'number': 1, 'start': 1, 'end': 501, 'size': 500,
             'filesystem': 'ext4', 'flags': 'boot'},
            {'number': 2, 'start': 501, 'end': 476940, 'size': 476439,
             'filesystem': '', 'flags': ''},
        ]
        execute_mock.return_value = (output, '')
        result = disk_utils.list_partitions('/dev/fake')
        self.assertEqual(expected, result)
        execute_mock.assert_called_once_with(
            'parted', '-s', '-m', '/dev/fake', 'unit', 'MiB', 'print',
            use_standard_locale=True, run_as_root=True)

    @mock.patch.object(disk_utils.LOG, 'warning', autospec=True)
    def test_incorrect(self, log_mock, execute_mock):
        output = """
BYT;
/dev/sda:500107862016B:scsi:512:4096:msdos:ATA HGST HTS725050A7:;
1:XX1076MiB:---:524MiB:ext4::boot;
"""
        execute_mock.return_value = (output, '')
        self.assertEqual([], disk_utils.list_partitions('/dev/fake'))
        self.assertEqual(1, log_mock.call_count)


@mock.patch.object(disk_partitioner.DiskPartitioner, 'commit', lambda _: None)
class WorkOnDiskTestCase(base.IronicLibTestCase):

    def setUp(self):
        super(WorkOnDiskTestCase, self).setUp()
        self.image_path = '/tmp/xyz/image'
        self.root_mb = 128
        self.swap_mb = 64
        self.ephemeral_mb = 0
        self.ephemeral_format = None
        self.configdrive_mb = 0
        self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
        self.dev = '/dev/fake'
        self.swap_part = '/dev/fake-part1'
        self.root_part = '/dev/fake-part2'

        self.mock_ibd_obj = mock.patch.object(
            disk_utils, 'is_block_device', autospec=True)
        self.mock_ibd = self.mock_ibd_obj.start()
        self.addCleanup(self.mock_ibd_obj.stop)
        self.mock_mp_obj = mock.patch.object(
            disk_utils, 'make_partitions', autospec=True)
        self.mock_mp = self.mock_mp_obj.start()
        self.addCleanup(self.mock_mp_obj.stop)
        self.mock_remlbl_obj = mock.patch.object(
            disk_utils, 'destroy_disk_metadata', autospec=True)
        self.mock_remlbl = self.mock_remlbl_obj.start()
        self.addCleanup(self.mock_remlbl_obj.stop)
        self.mock_mp.return_value = {'swap': self.swap_part,
                                     'root': self.root_part}

    def test_no_root_partition(self):
        self.mock_ibd.return_value = False
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils.work_on_disk, self.dev, self.root_mb,
                          self.swap_mb, self.ephemeral_mb,
                          self.ephemeral_format, self.image_path,
                          self.node_uuid)
        self.mock_ibd.assert_called_once_with(self.root_part)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, self.ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=True,
                                             boot_option="netboot",
                                             boot_mode="bios",
                                             disk_label=None,
                                             cpu_arch="")

    def test_no_swap_partition(self):
        self.mock_ibd.side_effect = iter([True, False])
        calls = [mock.call(self.root_part),
                 mock.call(self.swap_part)]
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils.work_on_disk, self.dev, self.root_mb,
                          self.swap_mb, self.ephemeral_mb,
                          self.ephemeral_format, self.image_path,
                          self.node_uuid)
        self.assertEqual(self.mock_ibd.call_args_list, calls)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, self.ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=True,
                                             boot_option="netboot",
                                             boot_mode="bios",
                                             disk_label=None,
                                             cpu_arch="")

    def test_no_ephemeral_partition(self):
        ephemeral_part = '/dev/fake-part1'
        swap_part = '/dev/fake-part2'
        root_part = '/dev/fake-part3'
        ephemeral_mb = 256
        ephemeral_format = 'exttest'

        self.mock_mp.return_value = {'ephemeral': ephemeral_part,
                                     'swap': swap_part,
                                     'root': root_part}
        self.mock_ibd.side_effect = iter([True, True, False])
        calls = [mock.call(root_part),
                 mock.call(swap_part),
                 mock.call(ephemeral_part)]
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils.work_on_disk, self.dev, self.root_mb,
                          self.swap_mb, ephemeral_mb, ephemeral_format,
                          self.image_path, self.node_uuid)
        self.assertEqual(self.mock_ibd.call_args_list, calls)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=True,
                                             boot_option="netboot",
                                             boot_mode="bios",
                                             disk_label=None,
                                             cpu_arch="")

    @mock.patch.object(utils, 'unlink_without_raise', autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive', autospec=True)
    def test_no_configdrive_partition(self, mock_configdrive, mock_unlink):
        mock_configdrive.return_value = (10, 'fake-path')
        swap_part = '/dev/fake-part1'
        configdrive_part = '/dev/fake-part2'
        root_part = '/dev/fake-part3'
        configdrive_url = 'http://1.2.3.4/cd'
        configdrive_mb = 10

        self.mock_mp.return_value = {'swap': swap_part,
                                     'configdrive': configdrive_part,
                                     'root': root_part}
        self.mock_ibd.side_effect = iter([True, True, False])
        calls = [mock.call(root_part),
                 mock.call(swap_part),
                 mock.call(configdrive_part)]
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils.work_on_disk, self.dev, self.root_mb,
                          self.swap_mb, self.ephemeral_mb,
                          self.ephemeral_format, self.image_path,
                          self.node_uuid, preserve_ephemeral=False,
                          configdrive=configdrive_url,
                          boot_option="netboot")
        self.assertEqual(self.mock_ibd.call_args_list, calls)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, self.ephemeral_mb,
                                             configdrive_mb, self.node_uuid,
                                             commit=True,
                                             boot_option="netboot",
                                             boot_mode="bios",
                                             disk_label=None,
                                             cpu_arch="")
        mock_unlink.assert_called_once_with('fake-path')

    @mock.patch.object(utils, 'mkfs', lambda fs, path, label=None: None)
    @mock.patch.object(disk_utils, 'block_uuid', lambda p: 'uuid')
    @mock.patch.object(disk_utils, 'populate_image', autospec=True)
    def test_without_image(self, mock_populate):
        ephemeral_part = '/dev/fake-part1'
        swap_part = '/dev/fake-part2'
        root_part = '/dev/fake-part3'
        ephemeral_mb = 256
        ephemeral_format = 'exttest'

        self.mock_mp.return_value = {'ephemeral': ephemeral_part,
                                     'swap': swap_part,
                                     'root': root_part}
        self.mock_ibd.return_value = True
        calls = [mock.call(root_part),
                 mock.call(swap_part),
                 mock.call(ephemeral_part)]
        res = disk_utils.work_on_disk(self.dev, self.root_mb,
                                      self.swap_mb, ephemeral_mb,
                                      ephemeral_format,
                                      None, self.node_uuid)
        self.assertEqual(self.mock_ibd.call_args_list, calls)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=True,
                                             boot_option="netboot",
                                             boot_mode="bios",
                                             disk_label=None,
                                             cpu_arch="")
        self.assertEqual(root_part, res['partitions']['root'])
        self.assertEqual('uuid', res['root uuid'])
        self.assertFalse(mock_populate.called)

    @mock.patch.object(utils, 'mkfs', lambda fs, path, label=None: None)
    @mock.patch.object(disk_utils, 'block_uuid', lambda p: 'uuid')
    @mock.patch.object(disk_utils, 'populate_image', lambda image_path,
                       root_path, conv_flags=None: None)
    def test_gpt_disk_label(self):
        ephemeral_part = '/dev/fake-part1'
        swap_part = '/dev/fake-part2'
        root_part = '/dev/fake-part3'
        ephemeral_mb = 256
        ephemeral_format = 'exttest'

        self.mock_mp.return_value = {'ephemeral': ephemeral_part,
                                     'swap': swap_part,
                                     'root': root_part}
        self.mock_ibd.return_value = True
        calls = [mock.call(root_part),
                 mock.call(swap_part),
                 mock.call(ephemeral_part)]
        disk_utils.work_on_disk(self.dev, self.root_mb,
                                self.swap_mb, ephemeral_mb, ephemeral_format,
                                self.image_path, self.node_uuid,
                                disk_label='gpt', conv_flags=None)
        self.assertEqual(self.mock_ibd.call_args_list, calls)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=True,
                                             boot_option="netboot",
                                             boot_mode="bios",
                                             disk_label='gpt',
                                             cpu_arch="")

    @mock.patch.object(disk_utils, 'block_uuid', autospec=True)
    @mock.patch.object(disk_utils, 'populate_image', autospec=True)
    @mock.patch.object(utils, 'mkfs', autospec=True)
    def test_uefi_localboot(self, mock_mkfs, mock_populate_image,
                            mock_block_uuid):
        """Test that we create a fat filesystem with UEFI localboot."""
        root_part = '/dev/fake-part1'
        efi_part = '/dev/fake-part2'
        self.mock_mp.return_value = {'root': root_part,
                                     'efi system partition': efi_part}
        self.mock_ibd.return_value = True
        mock_ibd_calls = [mock.call(root_part),
                          mock.call(efi_part)]

        disk_utils.work_on_disk(self.dev, self.root_mb,
                                self.swap_mb, self.ephemeral_mb,
                                self.ephemeral_format,
                                self.image_path, self.node_uuid,
                                boot_option="local", boot_mode="uefi")

        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, self.ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=True,
                                             boot_option="local",
                                             boot_mode="uefi",
                                             disk_label=None,
                                             cpu_arch="")
        self.assertEqual(self.mock_ibd.call_args_list, mock_ibd_calls)
        mock_mkfs.assert_called_once_with(fs='vfat', path=efi_part,
                                          label='efi-part')
        mock_populate_image.assert_called_once_with(self.image_path,
                                                    root_part, conv_flags=None)
        mock_block_uuid.assert_any_call(root_part)
        mock_block_uuid.assert_any_call(efi_part)

    @mock.patch.object(disk_utils, 'block_uuid', autospec=True)
    @mock.patch.object(disk_utils, 'populate_image', autospec=True)
    @mock.patch.object(utils, 'mkfs', autospec=True)
    def test_preserve_ephemeral(self, mock_mkfs, mock_populate_image,
                                mock_block_uuid):
        """Test that ephemeral partition doesn't get overwritten."""
        ephemeral_part = '/dev/fake-part1'
        root_part = '/dev/fake-part2'
        ephemeral_mb = 256
        ephemeral_format = 'exttest'

        self.mock_mp.return_value = {'ephemeral': ephemeral_part,
                                     'root': root_part}
        self.mock_ibd.return_value = True
        calls = [mock.call(root_part),
                 mock.call(ephemeral_part)]
        disk_utils.work_on_disk(self.dev, self.root_mb,
                                self.swap_mb, ephemeral_mb, ephemeral_format,
                                self.image_path, self.node_uuid,
                                preserve_ephemeral=True)
        self.assertEqual(self.mock_ibd.call_args_list, calls)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=False,
                                             boot_option="netboot",
                                             boot_mode="bios",
                                             disk_label=None,
                                             cpu_arch="")
        self.assertFalse(mock_mkfs.called)

    @mock.patch.object(disk_utils, 'block_uuid', autospec=True)
    @mock.patch.object(disk_utils, 'populate_image', autospec=True)
    @mock.patch.object(utils, 'mkfs', autospec=True)
    def test_ppc64le_prep_part(self, mock_mkfs, mock_populate_image,
                               mock_block_uuid):
        """Test that PReP partition uuid is returned."""
        prep_part = '/dev/fake-part1'
        root_part = '/dev/fake-part2'

        self.mock_mp.return_value = {'PReP Boot partition': prep_part,
                                     'root': root_part}
        self.mock_ibd.return_vaue = True
        calls = [mock.call(root_part),
                 mock.call(prep_part)]
        disk_utils.work_on_disk(self.dev, self.root_mb,
                                self.swap_mb, self.ephemeral_mb,
                                self.ephemeral_format, self.image_path,
                                self.node_uuid, boot_option="local",
                                cpu_arch='ppc64le')
        self.assertEqual(self.mock_ibd.call_args_list, calls)
        self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
                                             self.swap_mb, self.ephemeral_mb,
                                             self.configdrive_mb,
                                             self.node_uuid, commit=True,
                                             boot_option="local",
                                             boot_mode="bios",
                                             disk_label=None,
                                             cpu_arch="ppc64le")
        self.assertFalse(mock_mkfs.called)

    @mock.patch.object(disk_utils, 'block_uuid', autospec=True)
    @mock.patch.object(disk_utils, 'populate_image', autospec=True)
    @mock.patch.object(utils, 'mkfs', autospec=True)
    def test_convert_to_sparse(self, mock_mkfs, mock_populate_image,
                               mock_block_uuid):
        ephemeral_part = '/dev/fake-part1'
        swap_part = '/dev/fake-part2'
        root_part = '/dev/fake-part3'
        ephemeral_mb = 256
        ephemeral_format = 'exttest'

        self.mock_mp.return_value = {'ephemeral': ephemeral_part,
                                     'swap': swap_part,
                                     'root': root_part}
        self.mock_ibd.return_value = True
        disk_utils.work_on_disk(self.dev, self.root_mb,
                                self.swap_mb, ephemeral_mb, ephemeral_format,
                                self.image_path, self.node_uuid,
                                disk_label='gpt', conv_flags='sparse')

        mock_populate_image.assert_called_once_with(self.image_path,
                                                    root_part,
                                                    conv_flags='sparse')


@mock.patch.object(utils, 'execute', autospec=True)
class MakePartitionsTestCase(base.IronicLibTestCase):

    def setUp(self):
        super(MakePartitionsTestCase, self).setUp()
        self.dev = 'fake-dev'
        self.root_mb = 1024
        self.swap_mb = 512
        self.ephemeral_mb = 0
        self.configdrive_mb = 0
        self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
        self.efi_size = CONF.disk_utils.efi_system_partition_size
        self.bios_size = CONF.disk_utils.bios_boot_partition_size

    def _get_parted_cmd(self, dev, label=None):
        if label is None:
            label = 'msdos'

        return ['parted', '-a', 'optimal', '-s', dev,
                '--', 'unit', 'MiB', 'mklabel', label]

    def _test_make_partitions(self, mock_exc, boot_option, boot_mode='bios',
                              disk_label=None, cpu_arch=""):
        mock_exc.return_value = ('', '')
        disk_utils.make_partitions(self.dev, self.root_mb, self.swap_mb,
                                   self.ephemeral_mb, self.configdrive_mb,
                                   self.node_uuid, boot_option=boot_option,
                                   boot_mode=boot_mode, disk_label=disk_label,
                                   cpu_arch=cpu_arch)

        if boot_option == "local" and boot_mode == "uefi":
            add_efi_sz = lambda x: str(x + self.efi_size)
            expected_mkpart = ['mkpart', 'primary', 'fat32', '1',
                               add_efi_sz(1),
                               'set', '1', 'boot', 'on',
                               'mkpart', 'primary', 'linux-swap',
                               add_efi_sz(1), add_efi_sz(513), 'mkpart',
                               'primary', '', add_efi_sz(513),
                               add_efi_sz(1537)]
        else:
            if boot_option == "local":
                if disk_label == "gpt":
                    if cpu_arch.startswith('ppc64'):
                        expected_mkpart = ['mkpart', 'primary', '', '1', '9',
                                           'set', '1', 'prep', 'on',
                                           'mkpart', 'primary', 'linux-swap',
                                           '9', '521', 'mkpart', 'primary',
                                           '', '521', '1545']
                    else:
                        add_bios_sz = lambda x: str(x + self.bios_size)
                        expected_mkpart = ['mkpart', 'primary', '', '1',
                                           add_bios_sz(1),
                                           'set', '1', 'bios_grub', 'on',
                                           'mkpart', 'primary', 'linux-swap',
                                           add_bios_sz(1), add_bios_sz(513),
                                           'mkpart', 'primary', '',
                                           add_bios_sz(513), add_bios_sz(1537)]
                elif cpu_arch.startswith('ppc64'):
                    expected_mkpart = ['mkpart', 'primary', '', '1', '9',
                                       'set', '1', 'boot', 'on',
                                       'set', '1', 'prep', 'on',
                                       'mkpart', 'primary', 'linux-swap',
                                       '9', '521', 'mkpart', 'primary',
                                       '', '521', '1545']
                else:
                    expected_mkpart = ['mkpart', 'primary', 'linux-swap', '1',
                                       '513', 'mkpart', 'primary', '', '513',
                                       '1537', 'set', '2', 'boot', 'on']
            else:
                expected_mkpart = ['mkpart', 'primary', 'linux-swap', '1',
                                   '513', 'mkpart', 'primary', '', '513',
                                   '1537']
        self.dev = 'fake-dev'
        parted_cmd = (self._get_parted_cmd(self.dev, disk_label) +
                      expected_mkpart)
        parted_call = mock.call(*parted_cmd, use_standard_locale=True,
                                run_as_root=True, check_exit_code=[0])
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        mock_exc.assert_has_calls([parted_call, fuser_call])

    def test_make_partitions(self, mock_exc):
        self._test_make_partitions(mock_exc, boot_option="netboot")

    def test_make_partitions_local_boot(self, mock_exc):
        self._test_make_partitions(mock_exc, boot_option="local")

    def test_make_partitions_local_boot_uefi(self, mock_exc):
        self._test_make_partitions(mock_exc, boot_option="local",
                                   boot_mode="uefi", disk_label="gpt")

    def test_make_partitions_local_boot_gpt_bios(self, mock_exc):
        self._test_make_partitions(mock_exc, boot_option="local",
                                   disk_label="gpt")

    def test_make_partitions_disk_label_gpt(self, mock_exc):
        self._test_make_partitions(mock_exc, boot_option="netboot",
                                   disk_label="gpt")

    def test_make_partitions_mbr_with_prep(self, mock_exc):
        self._test_make_partitions(mock_exc, boot_option="local",
                                   disk_label="msdos", cpu_arch="ppc64le")

    def test_make_partitions_gpt_with_prep(self, mock_exc):
        self._test_make_partitions(mock_exc, boot_option="local",
                                   disk_label="gpt", cpu_arch="ppc64le")

    def test_make_partitions_with_ephemeral(self, mock_exc):
        self.ephemeral_mb = 2048
        expected_mkpart = ['mkpart', 'primary', '', '1', '2049',
                           'mkpart', 'primary', 'linux-swap', '2049', '2561',
                           'mkpart', 'primary', '', '2561', '3585']
        self.dev = 'fake-dev'
        cmd = self._get_parted_cmd(self.dev) + expected_mkpart
        mock_exc.return_value = ('', '')
        disk_utils.make_partitions(self.dev, self.root_mb, self.swap_mb,
                                   self.ephemeral_mb, self.configdrive_mb,
                                   self.node_uuid)

        parted_call = mock.call(*cmd, use_standard_locale=True,
                                run_as_root=True, check_exit_code=[0])
        mock_exc.assert_has_calls([parted_call])

    def test_make_partitions_with_iscsi_device(self, mock_exc):
        self.ephemeral_mb = 2048
        expected_mkpart = ['mkpart', 'primary', '', '1', '2049',
                           'mkpart', 'primary', 'linux-swap', '2049', '2561',
                           'mkpart', 'primary', '', '2561', '3585']
        self.dev = '/dev/iqn.2008-10.org.openstack:%s.fake-9' % self.node_uuid
        ep = '/dev/iqn.2008-10.org.openstack:%s.fake-9-part1' % self.node_uuid
        swap = ('/dev/iqn.2008-10.org.openstack:%s.fake-9-part2'
                % self.node_uuid)
        root = ('/dev/iqn.2008-10.org.openstack:%s.fake-9-part3'
                % self.node_uuid)
        expected_result = {'ephemeral': ep,
                           'swap': swap,
                           'root': root}
        cmd = self._get_parted_cmd(self.dev) + expected_mkpart
        mock_exc.return_value = ('', '')
        result = disk_utils.make_partitions(
            self.dev, self.root_mb, self.swap_mb, self.ephemeral_mb,
            self.configdrive_mb, self.node_uuid)

        parted_call = mock.call(*cmd, use_standard_locale=True,
                                run_as_root=True, check_exit_code=[0])
        mock_exc.assert_has_calls([parted_call])
        self.assertEqual(expected_result, result)

    def test_make_partitions_with_nvme_device(self, mock_exc):
        self.ephemeral_mb = 2048
        expected_mkpart = ['mkpart', 'primary', '', '1', '2049',
                           'mkpart', 'primary', 'linux-swap', '2049', '2561',
                           'mkpart', 'primary', '', '2561', '3585']
        self.dev = '/dev/nvmefake-9'
        ep = '/dev/nvmefake-9p1'
        swap = '/dev/nvmefake-9p2'
        root = '/dev/nvmefake-9p3'
        expected_result = {'ephemeral': ep,
                           'swap': swap,
                           'root': root}
        cmd = self._get_parted_cmd(self.dev) + expected_mkpart
        mock_exc.return_value = ('', '')
        result = disk_utils.make_partitions(
            self.dev, self.root_mb, self.swap_mb, self.ephemeral_mb,
            self.configdrive_mb, self.node_uuid)

        parted_call = mock.call(*cmd, use_standard_locale=True,
                                run_as_root=True, check_exit_code=[0])
        mock_exc.assert_has_calls([parted_call])
        self.assertEqual(expected_result, result)

    def test_make_partitions_with_local_device(self, mock_exc):
        self.ephemeral_mb = 2048
        expected_mkpart = ['mkpart', 'primary', '', '1', '2049',
                           'mkpart', 'primary', 'linux-swap', '2049', '2561',
                           'mkpart', 'primary', '', '2561', '3585']
        self.dev = 'fake-dev'
        expected_result = {'ephemeral': 'fake-dev1',
                           'swap': 'fake-dev2',
                           'root': 'fake-dev3'}
        cmd = self._get_parted_cmd(self.dev) + expected_mkpart
        mock_exc.return_value = ('', '')
        result = disk_utils.make_partitions(
            self.dev, self.root_mb, self.swap_mb, self.ephemeral_mb,
            self.configdrive_mb, self.node_uuid)

        parted_call = mock.call(*cmd, use_standard_locale=True,
                                run_as_root=True, check_exit_code=[0])
        mock_exc.assert_has_calls([parted_call])
        self.assertEqual(expected_result, result)


@mock.patch.object(utils, 'execute', autospec=True, return_value=('', ''))
class DestroyMetaDataTestCase(base.IronicLibTestCase):

    def setUp(self):
        super(DestroyMetaDataTestCase, self).setUp()
        self.dev = 'fake-dev'
        self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"

    def test_destroy_disk_metadata(self, mock_exec):
        expected_calls = [mock.call('wipefs', '--force', '--all', 'fake-dev',
                                    run_as_root=True,
                                    use_standard_locale=True),
                          mock.call('sgdisk', '-Z', 'fake-dev',
                                    run_as_root=True,
                                    use_standard_locale=True),
                          mock.call('fuser', self.dev,
                                    check_exit_code=[0, 1],
                                    run_as_root=True)]
        disk_utils.destroy_disk_metadata(self.dev, self.node_uuid)
        mock_exec.assert_has_calls(expected_calls)

    def test_destroy_disk_metadata_wipefs_fail(self, mock_exec):
        mock_exec.side_effect = processutils.ProcessExecutionError

        expected_call = [mock.call('wipefs', '--force', '--all', 'fake-dev',
                                   run_as_root=True,
                                   use_standard_locale=True)]
        self.assertRaises(processutils.ProcessExecutionError,
                          disk_utils.destroy_disk_metadata,
                          self.dev,
                          self.node_uuid)
        mock_exec.assert_has_calls(expected_call)

    def test_destroy_disk_metadata_sgdisk_fail(self, mock_exec):
        expected_calls = [mock.call('wipefs', '--force', '--all', 'fake-dev',
                                    run_as_root=True,
                                    use_standard_locale=True),
                          mock.call('sgdisk', '-Z', 'fake-dev',
                                    run_as_root=True,
                                    use_standard_locale=True)]
        mock_exec.side_effect = [(None, None),
                                 processutils.ProcessExecutionError()]
        self.assertRaises(processutils.ProcessExecutionError,
                          disk_utils.destroy_disk_metadata,
                          self.dev,
                          self.node_uuid)
        mock_exec.assert_has_calls(expected_calls)

    def test_destroy_disk_metadata_wipefs_not_support_force(self, mock_exec):
        mock_exec.side_effect = iter(
            [processutils.ProcessExecutionError(description='--force'),
             (None, None),
             (None, None),
             ('', '')])

        expected_call = [mock.call('wipefs', '--force', '--all', 'fake-dev',
                                   run_as_root=True,
                                   use_standard_locale=True),
                         mock.call('wipefs', '--all', 'fake-dev',
                                   run_as_root=True,
                                   use_standard_locale=True)]
        disk_utils.destroy_disk_metadata(self.dev, self.node_uuid)
        mock_exec.assert_has_calls(expected_call)


@mock.patch.object(utils, 'execute', autospec=True)
class GetDeviceBlockSizeTestCase(base.IronicLibTestCase):

    def setUp(self):
        super(GetDeviceBlockSizeTestCase, self).setUp()
        self.dev = 'fake-dev'
        self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"

    def test_get_dev_block_size(self, mock_exec):
        mock_exec.return_value = ("64", "")
        expected_call = [mock.call('blockdev', '--getsz', self.dev,
                                   run_as_root=True, check_exit_code=[0])]
        disk_utils.get_dev_block_size(self.dev)
        mock_exec.assert_has_calls(expected_call)


@mock.patch.object(disk_utils, 'dd', autospec=True)
@mock.patch.object(disk_utils, 'qemu_img_info', autospec=True)
@mock.patch.object(disk_utils, 'convert_image', autospec=True)
class PopulateImageTestCase(base.IronicLibTestCase):

    def test_populate_raw_image(self, mock_cg, mock_qinfo, mock_dd):
        type(mock_qinfo.return_value).file_format = mock.PropertyMock(
            return_value='raw')
        disk_utils.populate_image('src', 'dst')
        mock_dd.assert_called_once_with('src', 'dst', conv_flags=None)
        self.assertFalse(mock_cg.called)

    def test_populate_raw_image_with_convert(self, mock_cg, mock_qinfo,
                                             mock_dd):
        type(mock_qinfo.return_value).file_format = mock.PropertyMock(
            return_value='raw')
        disk_utils.populate_image('src', 'dst', conv_flags='sparse')
        mock_dd.assert_called_once_with('src', 'dst', conv_flags='sparse')
        self.assertFalse(mock_cg.called)

    def test_populate_qcow2_image(self, mock_cg, mock_qinfo, mock_dd):
        type(mock_qinfo.return_value).file_format = mock.PropertyMock(
            return_value='qcow2')
        disk_utils.populate_image('src', 'dst')
        mock_cg.assert_called_once_with('src', 'dst', 'raw', True)
        self.assertFalse(mock_dd.called)


@mock.patch.object(utils, 'wait_for_disk_to_become_available', lambda *_: None)
@mock.patch.object(disk_utils, 'is_block_device', lambda d: True)
@mock.patch.object(disk_utils, 'block_uuid', lambda p: 'uuid')
@mock.patch.object(disk_utils, 'dd', lambda *_: None)
@mock.patch.object(disk_utils, 'convert_image', lambda *_: None)
@mock.patch.object(utils, 'mkfs', lambda fs, path, label=None: None)
# NOTE(dtantsur): destroy_disk_metadata resets file size, disabling it
@mock.patch.object(disk_utils, 'destroy_disk_metadata', lambda *_: None)
class RealFilePartitioningTestCase(base.IronicLibTestCase):
    """This test applies some real-world partitioning scenario to a file.

    This test covers the whole partitioning, mocking everything not possible
    on a file. That helps us assure, that we do all partitioning math properly
    and also conducts integration testing of DiskPartitioner.
    """

    # Allow calls to utils.execute() and related functions
    block_execute = False

    def setUp(self):
        super(RealFilePartitioningTestCase, self).setUp()
        # NOTE(dtantsur): no parted utility on gate-ironic-python26
        try:
            utils.execute('parted', '--version')
        except OSError as exc:
            self.skipTest('parted utility was not found: %s' % exc)
        self.file = tempfile.NamedTemporaryFile(delete=False)
        # NOTE(ifarkas): the file needs to be closed, so fuser won't report
        #                any usage
        self.file.close()
        # NOTE(dtantsur): 20 MiB file with zeros
        utils.execute('dd', 'if=/dev/zero', 'of=%s' % self.file.name,
                      'bs=1', 'count=0', 'seek=20MiB')

    @staticmethod
    def _run_without_root(func, *args, **kwargs):
        """Make sure root is not required when using utils.execute."""
        real_execute = utils.execute

        def fake_execute(*cmd, **kwargs):
            kwargs['run_as_root'] = False
            return real_execute(*cmd, **kwargs)

        with mock.patch.object(utils, 'execute', fake_execute):
            return func(*args, **kwargs)

    def test_different_sizes(self):
        # NOTE(dtantsur): Keep this list in order with expected partitioning
        fields = ['ephemeral_mb', 'swap_mb', 'root_mb']
        variants = ((0, 0, 12), (4, 2, 8), (0, 4, 10), (5, 0, 10))
        for variant in variants:
            kwargs = dict(zip(fields, variant))
            self._run_without_root(disk_utils.work_on_disk,
                                   self.file.name, ephemeral_format='ext4',
                                   node_uuid='', image_path='path', **kwargs)
            part_table = self._run_without_root(
                disk_utils.list_partitions, self.file.name)
            for part, expected_size in zip(part_table, filter(None, variant)):
                self.assertEqual(expected_size, part['size'],
                                 "comparison failed for %s" % list(variant))

    def test_whole_disk(self):
        # 6 MiB ephemeral + 3 MiB swap + 9 MiB root + 1 MiB for MBR
        # + 1 MiB MAGIC == 20 MiB whole disk
        # TODO(dtantsur): figure out why we need 'magic' 1 more MiB
        # and why the is different on Ubuntu and Fedora (see below)
        self._run_without_root(disk_utils.work_on_disk, self.file.name,
                               root_mb=9, ephemeral_mb=6, swap_mb=3,
                               ephemeral_format='ext4', node_uuid='',
                               image_path='path')
        part_table = self._run_without_root(
            disk_utils.list_partitions, self.file.name)
        sizes = [part['size'] for part in part_table]
        # NOTE(dtantsur): parted in Ubuntu 12.04 will occupy the last MiB,
        # parted in Fedora 20 won't - thus two possible variants for last part
        self.assertEqual([6, 3], sizes[:2],
                         "unexpected partitioning %s" % part_table)
        self.assertIn(sizes[2], (9, 10))


@mock.patch.object(shutil, 'copyfileobj', autospec=True)
@mock.patch.object(requests, 'get', autospec=True)
class GetConfigdriveTestCase(base.IronicLibTestCase):

    @mock.patch.object(gzip, 'GzipFile', autospec=True)
    def test_get_configdrive(self, mock_gzip, mock_requests, mock_copy):
        mock_requests.return_value = mock.MagicMock(content='Zm9vYmFy')
        tempdir = tempfile.mkdtemp()
        (size, path) = disk_utils._get_configdrive('http://1.2.3.4/cd',
                                                   'fake-node-uuid',
                                                   tempdir=tempdir)
        self.assertTrue(path.startswith(tempdir))
        mock_requests.assert_called_once_with('http://1.2.3.4/cd')
        mock_gzip.assert_called_once_with('configdrive', 'rb',
                                          fileobj=mock.ANY)
        mock_copy.assert_called_once_with(mock.ANY, mock.ANY)

    @mock.patch.object(gzip, 'GzipFile', autospec=True)
    def test_get_configdrive_base64_string(self, mock_gzip, mock_requests,
                                           mock_copy):
        disk_utils._get_configdrive('Zm9vYmFy', 'fake-node-uuid')
        self.assertFalse(mock_requests.called)
        mock_gzip.assert_called_once_with('configdrive', 'rb',
                                          fileobj=mock.ANY)
        mock_copy.assert_called_once_with(mock.ANY, mock.ANY)

    def test_get_configdrive_bad_url(self, mock_requests, mock_copy):
        mock_requests.side_effect = requests.exceptions.RequestException
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils._get_configdrive,
                          'http://1.2.3.4/cd', 'fake-node-uuid')
        self.assertFalse(mock_copy.called)

    @mock.patch.object(base64, 'decode_as_bytes', autospec=True)
    def test_get_configdrive_base64_error(self, mock_b64, mock_requests,
                                          mock_copy):
        mock_b64.side_effect = TypeError
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils._get_configdrive,
                          'malformed', 'fake-node-uuid')
        mock_b64.assert_called_once_with('malformed')
        self.assertFalse(mock_copy.called)

    @mock.patch.object(gzip, 'GzipFile', autospec=True)
    def test_get_configdrive_gzip_error(self, mock_gzip, mock_requests,
                                        mock_copy):
        mock_requests.return_value = mock.MagicMock(content='Zm9vYmFy')
        mock_copy.side_effect = IOError
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils._get_configdrive,
                          'http://1.2.3.4/cd', 'fake-node-uuid')
        mock_requests.assert_called_once_with('http://1.2.3.4/cd')
        mock_gzip.assert_called_once_with('configdrive', 'rb',
                                          fileobj=mock.ANY)
        mock_copy.assert_called_once_with(mock.ANY, mock.ANY)


@mock.patch('time.sleep', lambda sec: None)
class OtherFunctionTestCase(base.IronicLibTestCase):

    @mock.patch.object(os, 'stat', autospec=True)
    @mock.patch.object(stat, 'S_ISBLK', autospec=True)
    def test_is_block_device_works(self, mock_is_blk, mock_os):
        device = '/dev/disk/by-path/ip-1.2.3.4:5678-iscsi-iqn.fake-lun-9'
        mock_is_blk.return_value = True
        mock_os().st_mode = 10000
        self.assertTrue(disk_utils.is_block_device(device))
        mock_is_blk.assert_called_once_with(mock_os().st_mode)

    @mock.patch.object(os, 'stat', autospec=True)
    def test_is_block_device_raises(self, mock_os):
        device = '/dev/disk/by-path/ip-1.2.3.4:5678-iscsi-iqn.fake-lun-9'
        mock_os.side_effect = OSError
        self.assertRaises(exception.InstanceDeployFailure,
                          disk_utils.is_block_device, device)
        mock_os.assert_has_calls([mock.call(device)] * 3)

    @mock.patch.object(imageutils, 'QemuImgInfo', autospec=True)
    @mock.patch.object(os.path, 'exists', return_value=False, autospec=True)
    def test_qemu_img_info_path_doesnt_exist(self, path_exists_mock,
                                             qemu_img_info_mock):
        disk_utils.qemu_img_info('noimg')
        path_exists_mock.assert_called_once_with('noimg')
        qemu_img_info_mock.assert_called_once_with()

    @mock.patch.object(utils, 'execute', return_value=('out', 'err'),
                       autospec=True)
    @mock.patch.object(imageutils, 'QemuImgInfo', autospec=True)
    @mock.patch.object(os.path, 'exists', return_value=True, autospec=True)
    def test_qemu_img_info_path_exists(self, path_exists_mock,
                                       qemu_img_info_mock, execute_mock):
        disk_utils.qemu_img_info('img')
        path_exists_mock.assert_called_once_with('img')
        execute_mock.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
                                             'qemu-img', 'info', 'img',
                                             prlimit=mock.ANY)
        qemu_img_info_mock.assert_called_once_with('out')

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_convert_image(self, execute_mock):
        disk_utils.convert_image('source', 'dest', 'out_format')
        execute_mock.assert_called_once_with('qemu-img', 'convert', '-O',
                                             'out_format', 'source', 'dest',
                                             run_as_root=False,
                                             prlimit=mock.ANY)

    @mock.patch.object(os.path, 'getsize', autospec=True)
    @mock.patch.object(disk_utils, 'qemu_img_info', autospec=True)
    def test_get_image_mb(self, mock_qinfo, mock_getsize):
        mb = 1024 * 1024

        mock_getsize.return_value = 0
        type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
            return_value=0)
        self.assertEqual(0, disk_utils.get_image_mb('x', False))
        self.assertEqual(0, disk_utils.get_image_mb('x', True))
        mock_getsize.return_value = 1
        type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
            return_value=1)
        self.assertEqual(1, disk_utils.get_image_mb('x', False))
        self.assertEqual(1, disk_utils.get_image_mb('x', True))
        mock_getsize.return_value = mb
        type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
            return_value=mb)
        self.assertEqual(1, disk_utils.get_image_mb('x', False))
        self.assertEqual(1, disk_utils.get_image_mb('x', True))
        mock_getsize.return_value = mb + 1
        type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
            return_value=mb + 1)
        self.assertEqual(2, disk_utils.get_image_mb('x', False))
        self.assertEqual(2, disk_utils.get_image_mb('x', True))

    def _test_count_mbr_partitions(self, output, mock_execute):
        mock_execute.return_value = (output, '')
        out = disk_utils.count_mbr_partitions('/dev/fake')
        mock_execute.assert_called_once_with('partprobe', '-d', '-s',
                                             '/dev/fake', run_as_root=True,
                                             use_standard_locale=True)
        return out

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_count_mbr_partitions(self, mock_execute):
        output = "/dev/fake: msdos partitions 1 2 3 <5 6>"
        pp, lp = self._test_count_mbr_partitions(output, mock_execute)
        self.assertEqual(3, pp)
        self.assertEqual(2, lp)

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_count_mbr_partitions_no_logical_partitions(self, mock_execute):
        output = "/dev/fake: msdos partitions 1 2"
        pp, lp = self._test_count_mbr_partitions(output, mock_execute)
        self.assertEqual(2, pp)
        self.assertEqual(0, lp)

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_count_mbr_partitions_wrong_partition_table(self, mock_execute):
        output = "/dev/fake: gpt partitions 1 2 3 4 5 6"
        mock_execute.return_value = (output, '')
        self.assertRaises(ValueError, disk_utils.count_mbr_partitions,
                          '/dev/fake')
        mock_execute.assert_called_once_with('partprobe', '-d', '-s',
                                             '/dev/fake', run_as_root=True,
                                             use_standard_locale=True)

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_block_uuid_fallback_to_uuid(self, mock_execute):
        mock_execute.side_effect = [('', ''),
                                    ('value', '')]
        self.assertEqual('value',
                         disk_utils.block_uuid('/dev/fake'))
        execute_calls = [
            mock.call('blkid', '-s', 'UUID', '-o', 'value',
                      '/dev/fake', check_exit_code=[0],
                      run_as_root=True),
            mock.call('blkid', '-s', 'PARTUUID', '-o', 'value',
                      '/dev/fake', check_exit_code=[0],
                      run_as_root=True)
        ]
        mock_execute.assert_has_calls(execute_calls)


@mock.patch.object(utils, 'execute', autospec=True)
class WholeDiskPartitionTestCases(base.IronicLibTestCase):

    def setUp(self):
        super(WholeDiskPartitionTestCases, self).setUp()
        self.dev = "/dev/fake"
        self.config_part_label = "config-2"
        self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"

    def test_get_partition_present(self, mock_execute):
        lsblk_output = 'NAME="fake12" LABEL="config-2"\n'
        part_result = '/dev/fake12'
        mock_execute.side_effect = [(None, ''), (lsblk_output, '')]
        result = disk_utils._get_labelled_partition(self.dev,
                                                    self.config_part_label,
                                                    self.node_uuid)
        self.assertEqual(part_result, result)
        execute_calls = [
            mock.call('partprobe', self.dev, run_as_root=True, attempts=10),
            mock.call('lsblk', '-Po', 'name,label', self.dev,
                      check_exit_code=[0, 1],
                      use_standard_locale=True, run_as_root=True)
        ]
        mock_execute.assert_has_calls(execute_calls)

    def test_get_partition_present_uppercase(self, mock_execute):
        lsblk_output = 'NAME="fake12" LABEL="CONFIG-2"\n'
        part_result = '/dev/fake12'
        mock_execute.side_effect = [(None, ''), (lsblk_output, '')]
        result = disk_utils._get_labelled_partition(self.dev,
                                                    self.config_part_label,
                                                    self.node_uuid)
        self.assertEqual(part_result, result)
        execute_calls = [
            mock.call('partprobe', self.dev, run_as_root=True, attempts=10),
            mock.call('lsblk', '-Po', 'name,label', self.dev,
                      check_exit_code=[0, 1],
                      use_standard_locale=True, run_as_root=True)
        ]
        mock_execute.assert_has_calls(execute_calls)

    def test_get_partition_absent(self, mock_execute):
        mock_execute.side_effect = [(None, ''),
                                    (None, '')]
        result = disk_utils._get_labelled_partition(self.dev,
                                                    self.config_part_label,
                                                    self.node_uuid)
        self.assertIsNone(result)
        execute_calls = [
            mock.call('partprobe', self.dev, run_as_root=True, attempts=10),
            mock.call('lsblk', '-Po', 'name,label', self.dev,
                      check_exit_code=[0, 1],
                      use_standard_locale=True, run_as_root=True)
        ]
        mock_execute.assert_has_calls(execute_calls)

    def test_get_partition_DeployFail_exc(self, mock_execute):
        label = 'config-2'
        lsblk_output = ('NAME="fake12" LABEL="%s"\n'
                        'NAME="fake13" LABEL="%s"\n' %
                        (label, label))
        mock_execute.side_effect = [(None, ''), (lsblk_output, '')]
        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'fake .*fake12 .*fake13',
                               disk_utils._get_labelled_partition, self.dev,
                               self.config_part_label, self.node_uuid)
        execute_calls = [
            mock.call('partprobe', self.dev, run_as_root=True, attempts=10),
            mock.call('lsblk', '-Po', 'name,label', self.dev,
                      check_exit_code=[0, 1],
                      use_standard_locale=True, run_as_root=True)
        ]
        mock_execute.assert_has_calls(execute_calls)

    @mock.patch.object(disk_utils.LOG, 'error', autospec=True)
    def test_get_partition_exc(self, mock_log, mock_execute):
        mock_execute.side_effect = processutils.ProcessExecutionError
        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Failed to retrieve partition labels',
                               disk_utils._get_labelled_partition, self.dev,
                               self.config_part_label, self.node_uuid)
        mock_execute.assert_called_once_with(
            'partprobe', self.dev, run_as_root=True, attempts=10)
        self.assertEqual(1, mock_log.call_count)

    def _test_is_disk_larger_than_max_size(self, mock_execute, blk_out):
        mock_execute.return_value = ('%s\n' % blk_out, '')
        result = disk_utils._is_disk_larger_than_max_size(self.dev,
                                                          self.node_uuid)
        mock_execute.assert_called_once_with('blockdev', '--getsize64',
                                             '/dev/fake', run_as_root=True,
                                             use_standard_locale=True)
        return result

    def test_is_disk_larger_than_max_size_false(self, mock_execute):
        blkid_out = "53687091200"
        ret = self._test_is_disk_larger_than_max_size(mock_execute,
                                                      blk_out=blkid_out)
        self.assertFalse(ret)

    def test_is_disk_larger_than_max_size_true(self, mock_execute):
        blkid_out = "4398046511104"
        ret = self._test_is_disk_larger_than_max_size(mock_execute,
                                                      blk_out=blkid_out)
        self.assertTrue(ret)

    @mock.patch.object(disk_utils.LOG, 'error', autospec=True)
    def test_is_disk_larger_than_max_size_exc(self, mock_log, mock_execute):
        mock_execute.side_effect = processutils.ProcessExecutionError
        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Failed to get size of disk',
                               disk_utils._is_disk_larger_than_max_size,
                               self.dev, self.node_uuid)
        mock_execute.assert_called_once_with('blockdev', '--getsize64',
                                             '/dev/fake', run_as_root=True,
                                             use_standard_locale=True)
        self.assertEqual(1, mock_log.call_count)

    def test__is_disk_gpt_partitioned_true(self, mock_execute):
        blkid_output = 'gpt\n'
        mock_execute.return_value = (blkid_output, '')
        result = disk_utils._is_disk_gpt_partitioned('/dev/fake',
                                                     self.node_uuid)
        self.assertTrue(result)
        mock_execute.assert_called_once_with('blkid', '-p', '-o', 'value',
                                             '-s', 'PTTYPE', '/dev/fake',
                                             use_standard_locale=True,
                                             run_as_root=True)

    def test_is_disk_gpt_partitioned_false(self, mock_execute):
        blkid_output = 'dos\n'
        mock_execute.return_value = (blkid_output, '')
        result = disk_utils._is_disk_gpt_partitioned('/dev/fake',
                                                     self.node_uuid)
        self.assertFalse(result)
        mock_execute.assert_called_once_with('blkid', '-p', '-o', 'value',
                                             '-s', 'PTTYPE', '/dev/fake',
                                             use_standard_locale=True,
                                             run_as_root=True)

    @mock.patch.object(disk_utils.LOG, 'error', autospec=True)
    def test_is_disk_gpt_partitioned_exc(self, mock_log, mock_execute):
        mock_execute.side_effect = processutils.ProcessExecutionError
        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Failed to retrieve partition table type',
                               disk_utils._is_disk_gpt_partitioned,
                               self.dev, self.node_uuid)
        mock_execute.assert_called_once_with('blkid', '-p', '-o', 'value',
                                             '-s', 'PTTYPE', '/dev/fake',
                                             use_standard_locale=True,
                                             run_as_root=True)
        self.assertEqual(1, mock_log.call_count)

    def test_fix_gpt_structs_fix_required(self, mock_execute):
        sgdisk_v_output = """
Problem: The secondary header's self-pointer indicates that it doesn't reside
at the end of the disk. If you've added a disk to a RAID array, use the 'e'
option on the experts' menu to adjust the secondary header's and partition
table's locations.

Identified 1 problems!
"""
        mock_execute.return_value = (sgdisk_v_output, '')
        execute_calls = [
            mock.call('sgdisk', '-v', '/dev/fake', run_as_root=True),
            mock.call('sgdisk', '-e', '/dev/fake', run_as_root=True)
        ]
        disk_utils._fix_gpt_structs('/dev/fake', self.node_uuid)
        mock_execute.assert_has_calls(execute_calls)

    def test_fix_gpt_structs_fix_not_required(self, mock_execute):
        mock_execute.return_value = ('', '')

        disk_utils._fix_gpt_structs('/dev/fake', self.node_uuid)
        mock_execute.assert_called_once_with('sgdisk', '-v', '/dev/fake',
                                             run_as_root=True)

    @mock.patch.object(disk_utils.LOG, 'error', autospec=True)
    def test_fix_gpt_structs_exc(self, mock_log, mock_execute):
        mock_execute.side_effect = processutils.ProcessExecutionError
        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Failed to fix GPT data structures on disk',
                               disk_utils._fix_gpt_structs,
                               self.dev, self.node_uuid)
        mock_execute.assert_called_once_with('sgdisk', '-v', '/dev/fake',
                                             run_as_root=True)
        self.assertEqual(1, mock_log.call_count)


class WholeDiskConfigDriveTestCases(base.IronicLibTestCase):

    def setUp(self):
        super(WholeDiskConfigDriveTestCases, self).setUp()
        self.dev = "/dev/fake"
        self.config_part_label = "config-2"
        self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"

    @mock.patch.object(utils, 'execute', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, 'dd',
                       autospec=True)
    @mock.patch.object(disk_utils, 'fix_gpt_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_fix_gpt_structs',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_gpt_partitioned',
                       autospec=True)
    @mock.patch.object(disk_utils, 'list_partitions',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def test_create_partition_exists(self, mock_get_configdrive,
                                     mock_get_labelled_partition,
                                     mock_list_partitions, mock_is_disk_gpt,
                                     mock_fix_gpt, mock_fix_gpt_partition,
                                     mock_dd, mock_unlink, mock_execute):
        config_url = 'http://1.2.3.4/cd'
        configdrive_part = '/dev/fake-part1'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 10

        mock_get_labelled_partition.return_value = configdrive_part
        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        disk_utils.create_config_drive_partition(self.node_uuid, self.dev,
                                                 config_url)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
        mock_get_labelled_partition.assert_called_with(self.dev,
                                                       self.config_part_label,
                                                       self.node_uuid)
        self.assertFalse(mock_list_partitions.called)
        self.assertFalse(mock_execute.called)
        self.assertFalse(mock_is_disk_gpt.called)
        self.assertFalse(mock_fix_gpt.called)
        mock_dd.assert_called_with(configdrive_file, configdrive_part)
        mock_unlink.assert_called_with(configdrive_file)

    @mock.patch.object(utils, 'execute', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, 'dd',
                       autospec=True)
    @mock.patch.object(disk_utils, 'fix_gpt_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_fix_gpt_structs',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_gpt_partitioned',
                       autospec=True)
    @mock.patch.object(disk_utils, 'list_partitions',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def test_create_partition_gpt(self, mock_get_configdrive,
                                  mock_get_labelled_partition,
                                  mock_list_partitions, mock_is_disk_gpt,
                                  mock_fix_gpt, mock_fix_gpt_partition,
                                  mock_dd, mock_unlink, mock_execute):
        config_url = 'http://1.2.3.4/cd'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 10

        initial_partitions = [{'end': 49152, 'number': 1, 'start': 1,
                               'flags': 'boot', 'filesystem': 'ext4',
                               'size': 49151},
                              {'end': 51099, 'number': 3, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 5, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046}]
        updated_partitions = [{'end': 49152, 'number': 1, 'start': 1,
                               'flags': 'boot', 'filesystem': 'ext4',
                               'size': 49151},
                              {'end': 51099, 'number': 3, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 4, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 5, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046}]

        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        mock_get_labelled_partition.return_value = None

        mock_is_disk_gpt.return_value = True
        mock_list_partitions.side_effect = [initial_partitions,
                                            updated_partitions]
        expected_part = '/dev/fake4'
        disk_utils.create_config_drive_partition(self.node_uuid, self.dev,
                                                 config_url)
        mock_execute.assert_has_calls([
            mock.call('sgdisk', '-n', '0:-64MB:0', self.dev,
                      run_as_root=True),
            mock.call('udevadm', 'settle'),
            mock.call('test', '-e', expected_part, attempts=15,
                      check_exit_code=[0], delay_on_retry=True)
        ])

        self.assertEqual(2, mock_list_partitions.call_count)
        mock_is_disk_gpt.assert_called_with(self.dev, self.node_uuid)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        self.assertFalse(mock_fix_gpt.called)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        mock_dd.assert_called_with(configdrive_file, expected_part)
        mock_unlink.assert_called_with(configdrive_file)

    @mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    @mock.patch.object(disk_utils.LOG, 'warning', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, 'dd',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_larger_than_max_size',
                       autospec=True)
    @mock.patch.object(disk_utils, 'fix_gpt_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_fix_gpt_structs',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_gpt_partitioned',
                       autospec=True)
    @mock.patch.object(disk_utils, 'list_partitions',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def _test_create_partition_mbr(self, mock_get_configdrive,
                                   mock_get_labelled_partition,
                                   mock_list_partitions,
                                   mock_is_disk_gpt, mock_fix_gpt,
                                   mock_fix_gpt_partition,
                                   mock_disk_exceeds, mock_dd,
                                   mock_unlink, mock_log, mock_execute,
                                   mock_count, disk_size_exceeds_max=False,
                                   is_iscsi_device=False,
                                   is_nvme_device=False):
        config_url = 'http://1.2.3.4/cd'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 10
        mock_disk_exceeds.return_value = disk_size_exceeds_max

        initial_partitions = [{'end': 49152, 'number': 1, 'start': 1,
                               'flags': 'boot', 'filesystem': 'ext4',
                               'size': 49151},
                              {'end': 51099, 'number': 3, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 5, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046}]
        updated_partitions = [{'end': 49152, 'number': 1, 'start': 1,
                               'flags': 'boot', 'filesystem': 'ext4',
                               'size': 49151},
                              {'end': 51099, 'number': 3, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 4, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 5, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046}]
        mock_list_partitions.side_effect = [initial_partitions,
                                            updated_partitions]
        # 2 primary partitions, 0 logical partitions
        mock_count.return_value = (2, 0)
        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        mock_get_labelled_partition.return_value = None
        mock_is_disk_gpt.return_value = False

        self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
        if is_iscsi_device:
            self.dev = ('/dev/iqn.2008-10.org.openstack:%s.fake' %
                        self.node_uuid)
            expected_part = '%s-part4' % self.dev
        elif is_nvme_device:
            self.dev = '/dev/nvmefake'
            expected_part = '%sp4' % self.dev
        else:
            expected_part = '/dev/fake4'

        disk_utils.create_config_drive_partition(self.node_uuid, self.dev,
                                                 config_url)
        mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
        if disk_size_exceeds_max:
            self.assertEqual(1, mock_log.call_count)
            parted_call = mock.call('parted', '-a', 'optimal', '-s',
                                    '--', self.dev, 'mkpart',
                                    'primary', 'fat32', 2097087,
                                    2097151, run_as_root=True)
        else:
            self.assertEqual(0, mock_log.call_count)
            parted_call = mock.call('parted', '-a', 'optimal', '-s',
                                    '--', self.dev, 'mkpart',
                                    'primary', 'fat32', '-64MiB',
                                    '-0', run_as_root=True)
        mock_execute.assert_has_calls([
            parted_call,
            mock.call('udevadm', 'settle'),
            mock.call('test', '-e', expected_part, attempts=15,
                      check_exit_code=[0], delay_on_retry=True)
        ])
        self.assertEqual(2, mock_list_partitions.call_count)
        mock_is_disk_gpt.assert_called_with(self.dev, self.node_uuid)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        mock_disk_exceeds.assert_called_with(self.dev, self.node_uuid)
        mock_dd.assert_called_with(configdrive_file, expected_part)
        mock_unlink.assert_called_with(configdrive_file)
        self.assertFalse(mock_fix_gpt.called)
        self.assertFalse(mock_fix_gpt.called)
        mock_count.assert_called_with(self.dev)

    def test__create_partition_mbr_disk_under_2TB(self):
        self._test_create_partition_mbr(disk_size_exceeds_max=False,
                                        is_iscsi_device=True,
                                        is_nvme_device=False)

    def test__create_partition_mbr_disk_under_2TB_nvme(self):
        self._test_create_partition_mbr(disk_size_exceeds_max=False,
                                        is_iscsi_device=False,
                                        is_nvme_device=True)

    def test__create_partition_mbr_disk_exceeds_2TB(self):
        self._test_create_partition_mbr(disk_size_exceeds_max=True,
                                        is_iscsi_device=False,
                                        is_nvme_device=False)

    def test__create_partition_mbr_disk_exceeds_2TB_nvme(self):
        self._test_create_partition_mbr(disk_size_exceeds_max=True,
                                        is_iscsi_device=False,
                                        is_nvme_device=True)

    @mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, 'dd',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_larger_than_max_size',
                       autospec=True)
    @mock.patch.object(disk_utils, 'fix_gpt_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_fix_gpt_structs',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_gpt_partitioned',
                       autospec=True)
    @mock.patch.object(disk_utils, 'list_partitions',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def test_create_partition_part_create_fail(self, mock_get_configdrive,
                                               mock_get_labelled_partition,
                                               mock_list_partitions,
                                               mock_is_disk_gpt, mock_fix_gpt,
                                               mock_fix_gpt_partition,
                                               mock_disk_exceeds, mock_dd,
                                               mock_unlink, mock_execute,
                                               mock_count):
        config_url = 'http://1.2.3.4/cd'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 10

        initial_partitions = [{'end': 49152, 'number': 1, 'start': 1,
                               'flags': 'boot', 'filesystem': 'ext4',
                               'size': 49151},
                              {'end': 51099, 'number': 3, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 5, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046}]
        updated_partitions = [{'end': 49152, 'number': 1, 'start': 1,
                               'flags': 'boot', 'filesystem': 'ext4',
                               'size': 49151},
                              {'end': 51099, 'number': 3, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 5, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046}]
        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        mock_get_labelled_partition.return_value = None
        mock_is_disk_gpt.return_value = False
        mock_disk_exceeds.return_value = False
        mock_list_partitions.side_effect = [initial_partitions,
                                            initial_partitions,
                                            updated_partitions]
        # 2 primary partitions, 0 logical partitions
        mock_count.return_value = (2, 0)

        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Disk partitioning failed on device',
                               disk_utils.create_config_drive_partition,
                               self.node_uuid, self.dev, config_url)

        mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
        mock_execute.assert_called_with('parted', '-a', 'optimal', '-s', '--',
                                        self.dev, 'mkpart', 'primary',
                                        'fat32', '-64MiB', '-0',
                                        run_as_root=True)
        self.assertEqual(2, mock_list_partitions.call_count)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        mock_is_disk_gpt.assert_called_with(self.dev, self.node_uuid)
        mock_disk_exceeds.assert_called_with(self.dev, self.node_uuid)
        self.assertFalse(mock_fix_gpt.called)
        self.assertFalse(mock_dd.called)
        mock_unlink.assert_called_with(configdrive_file)
        mock_count.assert_called_once_with(self.dev)

    @mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, 'dd',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_larger_than_max_size',
                       autospec=True)
    @mock.patch.object(disk_utils, 'fix_gpt_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_fix_gpt_structs',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_gpt_partitioned',
                       autospec=True)
    @mock.patch.object(disk_utils, 'list_partitions',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def test_create_partition_part_create_exc(self, mock_get_configdrive,
                                              mock_get_labelled_partition,
                                              mock_list_partitions,
                                              mock_is_disk_gpt, mock_fix_gpt,
                                              mock_fix_gpt_partition,
                                              mock_disk_exceeds, mock_dd,
                                              mock_unlink, mock_execute,
                                              mock_count):
        config_url = 'http://1.2.3.4/cd'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 10

        initial_partitions = [{'end': 49152, 'number': 1, 'start': 1,
                               'flags': 'boot', 'filesystem': 'ext4',
                               'size': 49151},
                              {'end': 51099, 'number': 3, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046},
                              {'end': 51099, 'number': 5, 'start': 49153,
                               'flags': '', 'filesystem': '', 'size': 2046}]
        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        mock_get_labelled_partition.return_value = None
        mock_is_disk_gpt.return_value = False
        mock_disk_exceeds.return_value = False
        mock_list_partitions.side_effect = [initial_partitions,
                                            initial_partitions]
        # 2 primary partitions, 0 logical partitions
        mock_count.return_value = (2, 0)

        mock_execute.side_effect = processutils.ProcessExecutionError

        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Failed to create config drive on disk',
                               disk_utils.create_config_drive_partition,
                               self.node_uuid, self.dev, config_url)

        mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
        mock_execute.assert_called_with('parted', '-a', 'optimal', '-s', '--',
                                        self.dev, 'mkpart', 'primary',
                                        'fat32', '-64MiB', '-0',
                                        run_as_root=True)
        self.assertEqual(1, mock_list_partitions.call_count)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        mock_is_disk_gpt.assert_called_with(self.dev, self.node_uuid)
        mock_disk_exceeds.assert_called_with(self.dev, self.node_uuid)
        self.assertFalse(mock_fix_gpt.called)
        self.assertFalse(mock_dd.called)
        mock_unlink.assert_called_with(configdrive_file)
        mock_count.assert_called_once_with(self.dev)

    @mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, 'dd',
                       autospec=True)
    @mock.patch.object(disk_utils, 'fix_gpt_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_fix_gpt_structs',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_gpt_partitioned',
                       autospec=True)
    @mock.patch.object(disk_utils, 'list_partitions',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def test_create_partition_num_parts_exceed(self, mock_get_configdrive,
                                               mock_get_labelled_partition,
                                               mock_list_partitions,
                                               mock_is_disk_gpt, mock_fix_gpt,
                                               mock_fix_gpt_partition,
                                               mock_dd, mock_unlink,
                                               mock_count):
        config_url = 'http://1.2.3.4/cd'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 10

        partitions = [{'end': 49152, 'number': 1, 'start': 1,
                       'flags': 'boot', 'filesystem': 'ext4',
                       'size': 49151},
                      {'end': 51099, 'number': 2, 'start': 49153,
                       'flags': '', 'filesystem': '', 'size': 2046},
                      {'end': 51099, 'number': 3, 'start': 49153,
                       'flags': '', 'filesystem': '', 'size': 2046},
                      {'end': 51099, 'number': 4, 'start': 49153,
                       'flags': '', 'filesystem': '', 'size': 2046}]
        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        mock_get_labelled_partition.return_value = None
        mock_is_disk_gpt.return_value = False
        mock_list_partitions.side_effect = [partitions, partitions]
        # 4 primary partitions, 0 logical partitions
        mock_count.return_value = (4, 0)

        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Config drive cannot be created for node',
                               disk_utils.create_config_drive_partition,
                               self.node_uuid, self.dev, config_url)

        mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
        self.assertEqual(1, mock_list_partitions.call_count)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        mock_is_disk_gpt.assert_called_with(self.dev, self.node_uuid)
        self.assertFalse(mock_fix_gpt.called)
        self.assertFalse(mock_dd.called)
        mock_unlink.assert_called_with(configdrive_file)
        mock_count.assert_called_once_with(self.dev)

    @mock.patch.object(utils, 'execute', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def test_create_partition_conf_drive_sz_exceed(self, mock_get_configdrive,
                                                   mock_get_labelled_partition,
                                                   mock_unlink, mock_execute):
        config_url = 'http://1.2.3.4/cd'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 65

        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        mock_get_labelled_partition.return_value = None

        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Config drive size exceeds maximum limit',
                               disk_utils.create_config_drive_partition,
                               self.node_uuid, self.dev, config_url)

        mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
        mock_unlink.assert_called_with(configdrive_file)

    @mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
    @mock.patch.object(utils, 'execute', autospec=True)
    @mock.patch.object(utils, 'unlink_without_raise',
                       autospec=True)
    @mock.patch.object(disk_utils, 'fix_gpt_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_is_disk_gpt_partitioned',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_labelled_partition',
                       autospec=True)
    @mock.patch.object(disk_utils, '_get_configdrive',
                       autospec=True)
    def test_create_partition_conf_drive_error_counting(
            self, mock_get_configdrive, mock_get_labelled_partition,
            mock_is_disk_gpt, mock_fix_gpt_partition,
            mock_unlink, mock_execute, mock_count):
        config_url = 'http://1.2.3.4/cd'
        configdrive_file = '/tmp/xyz'
        configdrive_mb = 10

        mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
        mock_get_labelled_partition.return_value = None
        mock_is_disk_gpt.return_value = False
        mock_count.side_effect = ValueError('Booooom')

        self.assertRaisesRegex(exception.InstanceDeployFailure,
                               'Failed to check the number of primary ',
                               disk_utils.create_config_drive_partition,
                               self.node_uuid, self.dev, config_url)

        mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
        mock_unlink.assert_called_with(configdrive_file)
        mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
        mock_is_disk_gpt.assert_called_with(self.dev, self.node_uuid)
        mock_count.assert_called_once_with(self.dev)




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\test_metrics.py
===========File Type===========
.py
===========File Content===========
# Copyright 2016 Rackspace Hosting
# 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.

import types

import mock
from oslo_utils import reflection

from ironic_lib import metrics as metricslib
from ironic_lib import metrics_utils
from ironic_lib.tests import base


METRICS = metrics_utils.get_metrics_logger(prefix='foo', backend='noop')


@METRICS.timer('testing1')
def timer_check(run, timer=None):
    pass


@METRICS.counter('testing2')
def counter_check(run, counter=None):
    pass


@METRICS.gauge('testing2')
def gauge_check(run, gauge=None):
    pass


class MockedMetricLogger(metricslib.MetricLogger):
    _gauge = mock.Mock(spec_set=types.FunctionType)
    _counter = mock.Mock(spec_set=types.FunctionType)
    _timer = mock.Mock(spec_set=types.FunctionType)


class TestMetricReflection(base.IronicLibTestCase):
    def test_timer_reflection(self):
        # Ensure our decorator is done correctly (six.wraps) and we can get the
        # arguments of our decorated function.
        expected = ['run', 'timer']
        signature = reflection.get_signature(timer_check)
        parameters = list(signature.parameters)
        self.assertEqual(expected, parameters)

    def test_counter_reflection(self):
        # Ensure our decorator is done correctly (six.wraps) and we can get the
        # arguments of our decorated function.
        expected = ['run', 'counter']
        signature = reflection.get_signature(counter_check)
        parameters = list(signature.parameters)
        self.assertEqual(expected, parameters)

    def test_gauge_reflection(self):
        # Ensure our decorator is done correctly (six.wraps) and we can get the
        # arguments of our decorated function.
        expected = ['run', 'gauge']
        signature = reflection.get_signature(gauge_check)
        parameters = list(signature.parameters)
        self.assertEqual(expected, parameters)


class TestMetricLogger(base.IronicLibTestCase):
    def setUp(self):
        super(TestMetricLogger, self).setUp()
        self.ml = MockedMetricLogger('prefix', '.')
        self.ml_no_prefix = MockedMetricLogger('', '.')
        self.ml_other_delim = MockedMetricLogger('prefix', '*')
        self.ml_default = MockedMetricLogger()

    def test_init(self):
        self.assertEqual(self.ml._prefix, 'prefix')
        self.assertEqual(self.ml._delimiter, '.')

        self.assertEqual(self.ml_no_prefix._prefix, '')
        self.assertEqual(self.ml_other_delim._delimiter, '*')
        self.assertEqual(self.ml_default._prefix, '')

    def test_get_metric_name(self):
        self.assertEqual(
            self.ml.get_metric_name('metric'),
            'prefix.metric')

        self.assertEqual(
            self.ml_no_prefix.get_metric_name('metric'),
            'metric')

        self.assertEqual(
            self.ml_other_delim.get_metric_name('metric'),
            'prefix*metric')

    def test_send_gauge(self):
        self.ml.send_gauge('prefix.metric', 10)
        self.ml._gauge.assert_called_once_with('prefix.metric', 10)

    def test_send_counter(self):
        self.ml.send_counter('prefix.metric', 10)
        self.ml._counter.assert_called_once_with(
            'prefix.metric', 10,
            sample_rate=None)
        self.ml._counter.reset_mock()

        self.ml.send_counter('prefix.metric', 10, sample_rate=1.0)
        self.ml._counter.assert_called_once_with(
            'prefix.metric', 10,
            sample_rate=1.0)
        self.ml._counter.reset_mock()

        self.ml.send_counter('prefix.metric', 10, sample_rate=0.0)
        self.assertFalse(self.ml._counter.called)

    def test_send_timer(self):
        self.ml.send_timer('prefix.metric', 10)
        self.ml._timer.assert_called_once_with('prefix.metric', 10)

    @mock.patch('ironic_lib.metrics._time', autospec=True)
    @mock.patch('ironic_lib.metrics.MetricLogger.send_timer', autospec=True)
    def test_decorator_timer(self, mock_timer, mock_time):
        mock_time.side_effect = [1, 43]

        @self.ml.timer('foo.bar.baz')
        def func(x):
            return x * x

        func(10)

        mock_timer.assert_called_once_with(self.ml, 'prefix.foo.bar.baz',
                                           42 * 1000)

    @mock.patch('ironic_lib.metrics.MetricLogger.send_counter', autospec=True)
    def test_decorator_counter(self, mock_counter):

        @self.ml.counter('foo.bar.baz')
        def func(x):
            return x * x

        func(10)

        mock_counter.assert_called_once_with(self.ml, 'prefix.foo.bar.baz', 1,
                                             sample_rate=None)

    @mock.patch('ironic_lib.metrics.MetricLogger.send_counter', autospec=True)
    def test_decorator_counter_sample_rate(self, mock_counter):

        @self.ml.counter('foo.bar.baz', sample_rate=0.5)
        def func(x):
            return x * x

        func(10)

        mock_counter.assert_called_once_with(self.ml, 'prefix.foo.bar.baz', 1,
                                             sample_rate=0.5)

    @mock.patch('ironic_lib.metrics.MetricLogger.send_gauge', autospec=True)
    def test_decorator_gauge(self, mock_gauge):
        @self.ml.gauge('foo.bar.baz')
        def func(x):
            return x

        func(10)

        mock_gauge.assert_called_once_with(self.ml, 'prefix.foo.bar.baz', 10)

    @mock.patch('ironic_lib.metrics._time', autospec=True)
    @mock.patch('ironic_lib.metrics.MetricLogger.send_timer', autospec=True)
    def test_context_mgr_timer(self, mock_timer, mock_time):
        mock_time.side_effect = [1, 43]

        with self.ml.timer('foo.bar.baz'):
            pass

        mock_timer.assert_called_once_with(self.ml, 'prefix.foo.bar.baz',
                                           42 * 1000)

    @mock.patch('ironic_lib.metrics.MetricLogger.send_counter', autospec=True)
    def test_context_mgr_counter(self, mock_counter):

        with self.ml.counter('foo.bar.baz'):
            pass

        mock_counter.assert_called_once_with(self.ml, 'prefix.foo.bar.baz', 1,
                                             sample_rate=None)

    @mock.patch('ironic_lib.metrics.MetricLogger.send_counter', autospec=True)
    def test_context_mgr_counter_sample_rate(self, mock_counter):

        with self.ml.counter('foo.bar.baz', sample_rate=0.5):
            pass

        mock_counter.assert_called_once_with(self.ml, 'prefix.foo.bar.baz', 1,
                                             sample_rate=0.5)




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\test_metrics_statsd.py
===========File Type===========
.py
===========File Content===========
# Copyright 2016 Rackspace Hosting
# 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.

import socket

import mock

from ironic_lib import metrics_statsd
from ironic_lib.tests import base


def connect(family=None, type=None, proto=None):
    """Dummy function to provide signature for autospec"""
    pass


class TestStatsdMetricLogger(base.IronicLibTestCase):
    def setUp(self):
        super(TestStatsdMetricLogger, self).setUp()
        self.ml = metrics_statsd.StatsdMetricLogger('prefix', '.', 'test-host',
                                                    4321)

    def test_init(self):
        self.assertEqual(self.ml._host, 'test-host')
        self.assertEqual(self.ml._port, 4321)
        self.assertEqual(self.ml._target, ('test-host', 4321))

    @mock.patch('ironic_lib.metrics_statsd.StatsdMetricLogger._send',
                autospec=True)
    def test_gauge(self, mock_send):
        self.ml._gauge('metric', 10)
        mock_send.assert_called_once_with(self.ml, 'metric', 10, 'g')

    @mock.patch('ironic_lib.metrics_statsd.StatsdMetricLogger._send',
                autospec=True)
    def test_counter(self, mock_send):
        self.ml._counter('metric', 10)
        mock_send.assert_called_once_with(self.ml, 'metric', 10, 'c',
                                          sample_rate=None)
        mock_send.reset_mock()

        self.ml._counter('metric', 10, sample_rate=1.0)
        mock_send.assert_called_once_with(self.ml, 'metric', 10, 'c',
                                          sample_rate=1.0)

    @mock.patch('ironic_lib.metrics_statsd.StatsdMetricLogger._send',
                autospec=True)
    def test_timer(self, mock_send):
        self.ml._timer('metric', 10)
        mock_send.assert_called_once_with(self.ml, 'metric', 10, 'ms')

    @mock.patch('socket.socket', autospec=connect)
    def test_open_socket(self, mock_socket_constructor):
        self.ml._open_socket()
        mock_socket_constructor.assert_called_once_with(
            socket.AF_INET,
            socket.SOCK_DGRAM)

    @mock.patch('socket.socket', autospec=connect)
    def test_send(self, mock_socket_constructor):
        mock_socket = mock.Mock()
        mock_socket_constructor.return_value = mock_socket

        self.ml._send('part1.part2', 2, 'type')
        mock_socket.sendto.assert_called_once_with(
            'part1.part2:2|type',
            ('test-host', 4321))
        mock_socket.close.assert_called_once_with()
        mock_socket.reset_mock()

        self.ml._send('part1.part2', 3.14159, 'type')
        mock_socket.sendto.assert_called_once_with(
            'part1.part2:3.14159|type',
            ('test-host', 4321))
        mock_socket.close.assert_called_once_with()
        mock_socket.reset_mock()

        self.ml._send('part1.part2', 5, 'type')
        mock_socket.sendto.assert_called_once_with(
            'part1.part2:5|type',
            ('test-host', 4321))
        mock_socket.close.assert_called_once_with()
        mock_socket.reset_mock()

        self.ml._send('part1.part2', 5, 'type', sample_rate=0.5)
        mock_socket.sendto.assert_called_once_with(
            'part1.part2:5|type@0.5',
            ('test-host', 4321))
        mock_socket.close.assert_called_once_with()




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\test_metrics_utils.py
===========File Type===========
.py
===========File Content===========
# Copyright 2016 Rackspace Hosting
# 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.

from oslo_config import cfg

from ironic_lib import exception
from ironic_lib import metrics as metricslib
from ironic_lib import metrics_statsd
from ironic_lib import metrics_utils
from ironic_lib.tests import base

CONF = cfg.CONF


class TestGetLogger(base.IronicLibTestCase):
    def setUp(self):
        super(TestGetLogger, self).setUp()

    def test_default_backend(self):
        metrics = metrics_utils.get_metrics_logger('foo')
        self.assertIsInstance(metrics, metricslib.NoopMetricLogger)

    def test_statsd_backend(self):
        CONF.set_override('backend', 'statsd', group='metrics')

        metrics = metrics_utils.get_metrics_logger('foo')
        self.assertIsInstance(metrics, metrics_statsd.StatsdMetricLogger)
        CONF.clear_override('backend', group='metrics')

    def test_nonexisting_backend(self):
        self.assertRaises(exception.InvalidMetricConfig,
                          metrics_utils.get_metrics_logger, 'foo', 'test')

    def test_numeric_prefix(self):
        self.assertRaises(exception.InvalidMetricConfig,
                          metrics_utils.get_metrics_logger, 1)

    def test_numeric_list_prefix(self):
        self.assertRaises(exception.InvalidMetricConfig,
                          metrics_utils.get_metrics_logger, (1, 2))

    def test_default_prefix(self):
        metrics = metrics_utils.get_metrics_logger()
        self.assertIsInstance(metrics, metricslib.NoopMetricLogger)
        self.assertEqual(metrics.get_metric_name("bar"), "bar")

    def test_prepend_host_backend(self):
        CONF.set_override('prepend_host', True, group='metrics')
        CONF.set_override('prepend_host_reverse', False, group='metrics')

        metrics = metrics_utils.get_metrics_logger(prefix='foo',
                                                   host="host.example.com")
        self.assertIsInstance(metrics, metricslib.NoopMetricLogger)
        self.assertEqual(metrics.get_metric_name("bar"),
                         "host.example.com.foo.bar")

        CONF.clear_override('prepend_host', group='metrics')
        CONF.clear_override('prepend_host_reverse', group='metrics')

    def test_prepend_global_prefix_host_backend(self):
        CONF.set_override('prepend_host', True, group='metrics')
        CONF.set_override('prepend_host_reverse', False, group='metrics')
        CONF.set_override('global_prefix', 'global_pre', group='metrics')

        metrics = metrics_utils.get_metrics_logger(prefix='foo',
                                                   host="host.example.com")
        self.assertIsInstance(metrics, metricslib.NoopMetricLogger)
        self.assertEqual(metrics.get_metric_name("bar"),
                         "global_pre.host.example.com.foo.bar")

        CONF.clear_override('prepend_host', group='metrics')
        CONF.clear_override('prepend_host_reverse', group='metrics')
        CONF.clear_override('global_prefix', group='metrics')

    def test_prepend_other_delim(self):
        metrics = metrics_utils.get_metrics_logger('foo', delimiter='*')
        self.assertIsInstance(metrics, metricslib.NoopMetricLogger)
        self.assertEqual(metrics.get_metric_name("bar"),
                         "foo*bar")

    def test_prepend_host_reverse_backend(self):
        CONF.set_override('prepend_host', True, group='metrics')
        CONF.set_override('prepend_host_reverse', True, group='metrics')

        metrics = metrics_utils.get_metrics_logger('foo',
                                                   host="host.example.com")
        self.assertIsInstance(metrics, metricslib.NoopMetricLogger)
        self.assertEqual(metrics.get_metric_name("bar"),
                         "com.example.host.foo.bar")

        CONF.clear_override('prepend_host', group='metrics')
        CONF.clear_override('prepend_host_reverse', group='metrics')




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\test_utils.py
===========File Type===========
.py
===========File Content===========
# Copyright 2011 Justin Santa Barbara
# Copyright 2012 Hewlett-Packard Development Company, L.P.
#
#    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.

import copy
import errno
import os
import os.path

import mock
from oslo_concurrency import processutils
from oslo_config import cfg

from ironic_lib import exception
from ironic_lib.tests import base
from ironic_lib import utils

CONF = cfg.CONF


class BareMetalUtilsTestCase(base.IronicLibTestCase):

    def test_unlink(self):
        with mock.patch.object(os, "unlink", autospec=True) as unlink_mock:
            unlink_mock.return_value = None
            utils.unlink_without_raise("/fake/path")
            unlink_mock.assert_called_once_with("/fake/path")

    def test_unlink_ENOENT(self):
        with mock.patch.object(os, "unlink", autospec=True) as unlink_mock:
            unlink_mock.side_effect = OSError(errno.ENOENT)
            utils.unlink_without_raise("/fake/path")
            unlink_mock.assert_called_once_with("/fake/path")


class ExecuteTestCase(base.IronicLibTestCase):
    # Allow calls to utils.execute() and related functions
    block_execute = False

    @mock.patch.object(processutils, 'execute', autospec=True)
    @mock.patch.object(os.environ, 'copy', return_value={}, autospec=True)
    def test_execute_use_standard_locale_no_env_variables(self, env_mock,
                                                          execute_mock):
        utils.execute('foo', use_standard_locale=True)
        execute_mock.assert_called_once_with('foo',
                                             env_variables={'LC_ALL': 'C'})

    @mock.patch.object(processutils, 'execute', autospec=True)
    def test_execute_use_standard_locale_with_env_variables(self,
                                                            execute_mock):
        utils.execute('foo', use_standard_locale=True,
                      env_variables={'foo': 'bar'})
        execute_mock.assert_called_once_with('foo',
                                             env_variables={'LC_ALL': 'C',
                                                            'foo': 'bar'})

    @mock.patch.object(processutils, 'execute', autospec=True)
    def test_execute_not_use_standard_locale(self, execute_mock):
        utils.execute('foo', use_standard_locale=False,
                      env_variables={'foo': 'bar'})
        execute_mock.assert_called_once_with('foo',
                                             env_variables={'foo': 'bar'})

    def test_execute_without_root_helper(self):
        CONF.set_override('root_helper', None, group='ironic_lib')
        with mock.patch.object(
                processutils, 'execute', autospec=True) as execute_mock:
            utils.execute('foo', run_as_root=False)
            execute_mock.assert_called_once_with('foo', run_as_root=False)

    def test_execute_without_root_helper_run_as_root(self):
        CONF.set_override('root_helper', None, group='ironic_lib')
        with mock.patch.object(
                processutils, 'execute', autospec=True) as execute_mock:
            utils.execute('foo', run_as_root=True)
            execute_mock.assert_called_once_with('foo', run_as_root=False)

    def test_execute_with_root_helper(self):
        with mock.patch.object(
                processutils, 'execute', autospec=True) as execute_mock:
            utils.execute('foo', run_as_root=False)
            execute_mock.assert_called_once_with('foo', run_as_root=False)

    def test_execute_with_root_helper_run_as_root(self):
        with mock.patch.object(
                processutils, 'execute', autospec=True) as execute_mock:
            utils.execute('foo', run_as_root=True)
            execute_mock.assert_called_once_with(
                'foo', run_as_root=True,
                root_helper=CONF.ironic_lib.root_helper)

    @mock.patch.object(utils, 'LOG', autospec=True)
    def _test_execute_with_log_stdout(self, log_mock, log_stdout=None):
        with mock.patch.object(
                processutils, 'execute', autospec=True) as execute_mock:
            execute_mock.return_value = ('stdout', 'stderr')
            if log_stdout is not None:
                utils.execute('foo', log_stdout=log_stdout)
            else:
                utils.execute('foo')
            execute_mock.assert_called_once_with('foo')
            name, args, kwargs = log_mock.debug.mock_calls[1]
            if log_stdout is False:
                self.assertEqual(2, log_mock.debug.call_count)
                self.assertNotIn('stdout', args[0])
            else:
                self.assertEqual(3, log_mock.debug.call_count)
                self.assertIn('stdout', args[0])

    def test_execute_with_log_stdout_default(self):
        self._test_execute_with_log_stdout()

    def test_execute_with_log_stdout_true(self):
        self._test_execute_with_log_stdout(log_stdout=True)

    def test_execute_with_log_stdout_false(self):
        self._test_execute_with_log_stdout(log_stdout=False)


class MkfsTestCase(base.IronicLibTestCase):

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_mkfs(self, execute_mock):
        utils.mkfs('ext4', '/my/block/dev')
        utils.mkfs('msdos', '/my/msdos/block/dev')
        utils.mkfs('swap', '/my/swap/block/dev')

        expected = [mock.call('mkfs', '-t', 'ext4', '-F', '/my/block/dev',
                              run_as_root=True,
                              use_standard_locale=True),
                    mock.call('mkfs', '-t', 'msdos', '/my/msdos/block/dev',
                              run_as_root=True,
                              use_standard_locale=True),
                    mock.call('mkswap', '/my/swap/block/dev',
                              run_as_root=True,
                              use_standard_locale=True)]
        self.assertEqual(expected, execute_mock.call_args_list)

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_mkfs_with_label(self, execute_mock):
        utils.mkfs('ext4', '/my/block/dev', 'ext4-vol')
        utils.mkfs('msdos', '/my/msdos/block/dev', 'msdos-vol')
        utils.mkfs('swap', '/my/swap/block/dev', 'swap-vol')

        expected = [mock.call('mkfs', '-t', 'ext4', '-F', '-L', 'ext4-vol',
                              '/my/block/dev', run_as_root=True,
                              use_standard_locale=True),
                    mock.call('mkfs', '-t', 'msdos', '-n', 'msdos-vol',
                              '/my/msdos/block/dev', run_as_root=True,
                              use_standard_locale=True),
                    mock.call('mkswap', '-L', 'swap-vol',
                              '/my/swap/block/dev', run_as_root=True,
                              use_standard_locale=True)]
        self.assertEqual(expected, execute_mock.call_args_list)

    @mock.patch.object(utils, 'execute', autospec=True,
                       side_effect=processutils.ProcessExecutionError(
                           stderr=os.strerror(errno.ENOENT)))
    def test_mkfs_with_unsupported_fs(self, execute_mock):
        self.assertRaises(exception.FileSystemNotSupported,
                          utils.mkfs, 'foo', '/my/block/dev')

    @mock.patch.object(utils, 'execute', autospec=True,
                       side_effect=processutils.ProcessExecutionError(
                           stderr='fake'))
    def test_mkfs_with_unexpected_error(self, execute_mock):
        self.assertRaises(processutils.ProcessExecutionError, utils.mkfs,
                          'ext4', '/my/block/dev', 'ext4-vol')


class IsHttpUrlTestCase(base.IronicLibTestCase):

    def test_is_http_url(self):
        self.assertTrue(utils.is_http_url('http://127.0.0.1'))
        self.assertTrue(utils.is_http_url('https://127.0.0.1'))
        self.assertTrue(utils.is_http_url('HTTP://127.1.2.3'))
        self.assertTrue(utils.is_http_url('HTTPS://127.3.2.1'))
        self.assertFalse(utils.is_http_url('Zm9vYmFy'))
        self.assertFalse(utils.is_http_url('11111111'))


class ParseRootDeviceTestCase(base.IronicLibTestCase):

    def test_parse_root_device_hints_without_operators(self):
        root_device = {
            'wwn': '123456', 'model': 'FOO model', 'size': 12345,
            'serial': 'foo-serial', 'vendor': 'foo VENDOR with space',
            'name': '/dev/sda', 'wwn_with_extension': '123456111',
            'wwn_vendor_extension': '111', 'rotational': True,
            'hctl': '1:0:0:0', 'by_path': '/dev/disk/by-path/1:0:0:0'}
        result = utils.parse_root_device_hints(root_device)
        expected = {
            'wwn': 's== 123456', 'model': 's== foo%20model',
            'size': '== 12345', 'serial': 's== foo-serial',
            'vendor': 's== foo%20vendor%20with%20space',
            'name': 's== /dev/sda', 'wwn_with_extension': 's== 123456111',
            'wwn_vendor_extension': 's== 111', 'rotational': True,
            'hctl': 's== 1%3A0%3A0%3A0',
            'by_path': 's== /dev/disk/by-path/1%3A0%3A0%3A0'}
        self.assertEqual(expected, result)

    def test_parse_root_device_hints_with_operators(self):
        root_device = {
            'wwn': 's== 123456', 'model': 's== foo MODEL', 'size': '>= 12345',
            'serial': 's!= foo-serial', 'vendor': 's== foo VENDOR with space',
            'name': '<or> /dev/sda <or> /dev/sdb',
            'wwn_with_extension': 's!= 123456111',
            'wwn_vendor_extension': 's== 111', 'rotational': True,
            'hctl': 's== 1:0:0:0', 'by_path': 's== /dev/disk/by-path/1:0:0:0'}

        # Validate strings being normalized
        expected = copy.deepcopy(root_device)
        expected['model'] = 's== foo%20model'
        expected['vendor'] = 's== foo%20vendor%20with%20space'
        expected['hctl'] = 's== 1%3A0%3A0%3A0'
        expected['by_path'] = 's== /dev/disk/by-path/1%3A0%3A0%3A0'

        result = utils.parse_root_device_hints(root_device)
        # The hints already contain the operators, make sure we keep it
        self.assertEqual(expected, result)

    def test_parse_root_device_hints_no_hints(self):
        result = utils.parse_root_device_hints({})
        self.assertIsNone(result)

    def test_parse_root_device_hints_convert_size(self):
        for size in (12345, '12345'):
            result = utils.parse_root_device_hints({'size': size})
            self.assertEqual({'size': '== 12345'}, result)

    def test_parse_root_device_hints_invalid_size(self):
        for value in ('not-int', -123, 0):
            self.assertRaises(ValueError, utils.parse_root_device_hints,
                              {'size': value})

    def test_parse_root_device_hints_int_or(self):
        expr = '<or> 123 <or> 456 <or> 789'
        result = utils.parse_root_device_hints({'size': expr})
        self.assertEqual({'size': expr}, result)

    def test_parse_root_device_hints_int_or_invalid(self):
        expr = '<or> 123 <or> non-int <or> 789'
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'size': expr})

    def test_parse_root_device_hints_string_or_space(self):
        expr = '<or> foo <or> foo bar <or> bar'
        expected = '<or> foo <or> foo%20bar <or> bar'
        result = utils.parse_root_device_hints({'model': expr})
        self.assertEqual({'model': expected}, result)

    def _parse_root_device_hints_convert_rotational(self, values,
                                                    expected_value):
        for value in values:
            result = utils.parse_root_device_hints({'rotational': value})
            self.assertEqual({'rotational': expected_value}, result)

    def test_parse_root_device_hints_convert_rotational(self):
        self._parse_root_device_hints_convert_rotational(
            (True, 'true', 'on', 'y', 'yes'), True)

        self._parse_root_device_hints_convert_rotational(
            (False, 'false', 'off', 'n', 'no'), False)

    def test_parse_root_device_hints_invalid_rotational(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'rotational': 'not-bool'})

    def test_parse_root_device_hints_invalid_wwn(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'wwn': 123})

    def test_parse_root_device_hints_invalid_wwn_with_extension(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'wwn_with_extension': 123})

    def test_parse_root_device_hints_invalid_wwn_vendor_extension(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'wwn_vendor_extension': 123})

    def test_parse_root_device_hints_invalid_model(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'model': 123})

    def test_parse_root_device_hints_invalid_serial(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'serial': 123})

    def test_parse_root_device_hints_invalid_vendor(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'vendor': 123})

    def test_parse_root_device_hints_invalid_name(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'name': 123})

    def test_parse_root_device_hints_invalid_hctl(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'hctl': 123})

    def test_parse_root_device_hints_invalid_by_path(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'by_path': 123})

    def test_parse_root_device_hints_non_existent_hint(self):
        self.assertRaises(ValueError, utils.parse_root_device_hints,
                          {'non-existent': 'foo'})

    def test_extract_hint_operator_and_values_single_value(self):
        expected = {'op': '>=', 'values': ['123']}
        self.assertEqual(
            expected, utils._extract_hint_operator_and_values(
                '>= 123', 'size'))

    def test_extract_hint_operator_and_values_multiple_values(self):
        expected = {'op': '<or>', 'values': ['123', '456', '789']}
        expr = '<or> 123 <or> 456 <or> 789'
        self.assertEqual(
            expected, utils._extract_hint_operator_and_values(expr, 'size'))

    def test_extract_hint_operator_and_values_multiple_values_space(self):
        expected = {'op': '<or>', 'values': ['foo', 'foo bar', 'bar']}
        expr = '<or> foo <or> foo bar <or> bar'
        self.assertEqual(
            expected, utils._extract_hint_operator_and_values(expr, 'model'))

    def test_extract_hint_operator_and_values_no_operator(self):
        expected = {'op': '', 'values': ['123']}
        self.assertEqual(
            expected, utils._extract_hint_operator_and_values('123', 'size'))

    def test_extract_hint_operator_and_values_empty_value(self):
        self.assertRaises(
            ValueError, utils._extract_hint_operator_and_values, '', 'size')

    def test_extract_hint_operator_and_values_integer(self):
        expected = {'op': '', 'values': ['123']}
        self.assertEqual(
            expected, utils._extract_hint_operator_and_values(123, 'size'))

    def test__append_operator_to_hints(self):
        root_device = {'serial': 'foo', 'size': 12345,
                       'model': 'foo model', 'rotational': True}
        expected = {'serial': 's== foo', 'size': '== 12345',
                    'model': 's== foo model', 'rotational': True}

        result = utils._append_operator_to_hints(root_device)
        self.assertEqual(expected, result)

    def test_normalize_hint_expression_or(self):
        expr = '<or> foo <or> foo bar <or> bar'
        expected = '<or> foo <or> foo%20bar <or> bar'
        result = utils._normalize_hint_expression(expr, 'model')
        self.assertEqual(expected, result)

    def test_normalize_hint_expression_in(self):
        expr = '<in> foo <in> foo bar <in> bar'
        expected = '<in> foo <in> foo%20bar <in> bar'
        result = utils._normalize_hint_expression(expr, 'model')
        self.assertEqual(expected, result)

    def test_normalize_hint_expression_op_space(self):
        expr = 's== test string with space'
        expected = 's== test%20string%20with%20space'
        result = utils._normalize_hint_expression(expr, 'model')
        self.assertEqual(expected, result)

    def test_normalize_hint_expression_op_no_space(self):
        expr = 's!= SpongeBob'
        expected = 's!= spongebob'
        result = utils._normalize_hint_expression(expr, 'model')
        self.assertEqual(expected, result)

    def test_normalize_hint_expression_no_op_space(self):
        expr = 'no operators'
        expected = 'no%20operators'
        result = utils._normalize_hint_expression(expr, 'model')
        self.assertEqual(expected, result)

    def test_normalize_hint_expression_no_op_no_space(self):
        expr = 'NoSpace'
        expected = 'nospace'
        result = utils._normalize_hint_expression(expr, 'model')
        self.assertEqual(expected, result)

    def test_normalize_hint_expression_empty_value(self):
        self.assertRaises(
            ValueError, utils._normalize_hint_expression, '', 'size')


class MatchRootDeviceTestCase(base.IronicLibTestCase):

    def setUp(self):
        super(MatchRootDeviceTestCase, self).setUp()
        self.devices = [
            {'name': '/dev/sda', 'size': 64424509440, 'model': 'ok model',
             'serial': 'fakeserial'},
            {'name': '/dev/sdb', 'size': 128849018880, 'model': 'big model',
             'serial': 'veryfakeserial', 'rotational': 'yes'},
            {'name': '/dev/sdc', 'size': 10737418240, 'model': 'small model',
             'serial': 'veryveryfakeserial', 'rotational': False},
        ]

    def test_match_root_device_hints_one_hint(self):
        root_device_hints = {'size': '>= 70'}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertEqual('/dev/sdb', dev['name'])

    def test_match_root_device_hints_rotational(self):
        root_device_hints = {'rotational': False}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertEqual('/dev/sdc', dev['name'])

    def test_match_root_device_hints_rotational_convert_devices_bool(self):
        root_device_hints = {'size': '>=100', 'rotational': True}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertEqual('/dev/sdb', dev['name'])

    def test_match_root_device_hints_multiple_hints(self):
        root_device_hints = {'size': '>= 50', 'model': 's==big model',
                             'serial': 's==veryfakeserial'}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertEqual('/dev/sdb', dev['name'])

    def test_match_root_device_hints_multiple_hints2(self):
        root_device_hints = {
            'size': '<= 20',
            'model': '<or> model 5 <or> foomodel <or> small model <or>',
            'serial': 's== veryveryfakeserial'}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertEqual('/dev/sdc', dev['name'])

    def test_match_root_device_hints_multiple_hints3(self):
        root_device_hints = {'rotational': False, 'model': '<in> small'}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertEqual('/dev/sdc', dev['name'])

    def test_match_root_device_hints_no_operators(self):
        root_device_hints = {'size': '120', 'model': 'big model',
                             'serial': 'veryfakeserial'}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertEqual('/dev/sdb', dev['name'])

    def test_match_root_device_hints_no_device_found(self):
        root_device_hints = {'size': '>=50', 'model': 's==foo'}
        dev = utils.match_root_device_hints(self.devices, root_device_hints)
        self.assertIsNone(dev)

    @mock.patch.object(utils.LOG, 'warning', autospec=True)
    def test_match_root_device_hints_empty_device_attribute(self, mock_warn):
        empty_dev = [{'name': '/dev/sda', 'model': ' '}]
        dev = utils.match_root_device_hints(empty_dev, {'model': 'foo'})
        self.assertIsNone(dev)
        self.assertTrue(mock_warn.called)


class WaitForDisk(base.IronicLibTestCase):

    def setUp(self):
        super(WaitForDisk, self).setUp()
        CONF.set_override('check_device_interval', .01,
                          group='disk_partitioner')
        CONF.set_override('check_device_max_retries', 2,
                          group='disk_partitioner')

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_wait_for_disk_to_become_available(self, mock_exc):
        mock_exc.return_value = ('', '')
        utils.wait_for_disk_to_become_available('fake-dev')
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        self.assertEqual(1, mock_exc.call_count)
        mock_exc.assert_has_calls([fuser_call])

    @mock.patch.object(utils, 'execute', autospec=True,
                       side_effect=processutils.ProcessExecutionError(
                           stderr='fake'))
    def test_wait_for_disk_to_become_available_no_fuser(self, mock_exc):
        self.assertRaises(exception.IronicException,
                          utils.wait_for_disk_to_become_available,
                          'fake-dev')
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        self.assertEqual(2, mock_exc.call_count)
        mock_exc.assert_has_calls([fuser_call, fuser_call])

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_wait_for_disk_to_become_available_device_in_use_psmisc(
            self, mock_exc):
        # Test that the device is not available. This version has the 'psmisc'
        # version of 'fuser' values for stdout and stderr.
        # NOTE(TheJulia): Looks like fuser returns the actual list of pids
        # in the stdout output, where as all other text is returned in
        # stderr.
        # The 'psmisc' version has a leading space character in stdout. The
        # filename is output to stderr
        mock_exc.side_effect = [(' 1234   ', 'fake-dev: '),
                                (' 15503  3919 15510 15511', 'fake-dev:')]
        expected_error = ('Processes with the following PIDs are '
                          'holding device fake-dev: 15503, 3919, 15510, '
                          '15511. Timed out waiting for completion.')
        self.assertRaisesRegex(
            exception.IronicException,
            expected_error,
            utils.wait_for_disk_to_become_available,
            'fake-dev')
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        self.assertEqual(2, mock_exc.call_count)
        mock_exc.assert_has_calls([fuser_call, fuser_call])

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_wait_for_disk_to_become_available_device_in_use_busybox(
            self, mock_exc):
        # Test that the device is not available. This version has the 'busybox'
        # version of 'fuser' values for stdout and stderr.
        # NOTE(TheJulia): Looks like fuser returns the actual list of pids
        # in the stdout output, where as all other text is returned in
        # stderr.
        # The 'busybox' version does not have a leading space character in
        # stdout. Also nothing is output to stderr.
        mock_exc.side_effect = [('1234', ''),
                                ('15503  3919 15510 15511', '')]
        expected_error = ('Processes with the following PIDs are '
                          'holding device fake-dev: 15503, 3919, 15510, '
                          '15511. Timed out waiting for completion.')
        self.assertRaisesRegex(
            exception.IronicException,
            expected_error,
            utils.wait_for_disk_to_become_available,
            'fake-dev')
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        self.assertEqual(2, mock_exc.call_count)
        mock_exc.assert_has_calls([fuser_call, fuser_call])

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_wait_for_disk_to_become_available_no_device(self, mock_exc):
        # NOTE(TheJulia): Looks like fuser returns the actual list of pids
        # in the stdout output, where as all other text is returned in
        # stderr.

        mock_exc.return_value = ('', 'Specified filename /dev/fake '
                                     'does not exist.')
        expected_error = ('Fuser exited with "Specified filename '
                          '/dev/fake does not exist." while checking '
                          'locks for device fake-dev. Timed out waiting '
                          'for completion.')
        self.assertRaisesRegex(
            exception.IronicException,
            expected_error,
            utils.wait_for_disk_to_become_available,
            'fake-dev')
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        self.assertEqual(2, mock_exc.call_count)
        mock_exc.assert_has_calls([fuser_call, fuser_call])

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_wait_for_disk_to_become_available_dev_becomes_avail_psmisc(
            self, mock_exc):
        # Test that initially device is not available but then becomes
        # available. This version has the 'psmisc' version of 'fuser' values
        # for stdout and stderr.
        # The 'psmisc' version has a leading space character in stdout. The
        # filename is output to stderr
        mock_exc.side_effect = [(' 1234   ', 'fake-dev: '),
                                ('', '')]
        utils.wait_for_disk_to_become_available('fake-dev')
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        self.assertEqual(2, mock_exc.call_count)
        mock_exc.assert_has_calls([fuser_call, fuser_call])

    @mock.patch.object(utils, 'execute', autospec=True)
    def test_wait_for_disk_to_become_available_dev_becomes_avail_busybox(
            self, mock_exc):
        # Test that initially device is not available but then becomes
        # available. This version has the 'busybox' version of 'fuser' values
        # for stdout and stderr.
        # The 'busybox' version does not have a leading space character in
        # stdout. Also nothing is output to stderr.
        mock_exc.side_effect = [('1234 5895', ''),
                                ('', '')]
        utils.wait_for_disk_to_become_available('fake-dev')
        fuser_cmd = ['fuser', 'fake-dev']
        fuser_call = mock.call(*fuser_cmd, run_as_root=True,
                               check_exit_code=[0, 1])
        self.assertEqual(2, mock_exc.call_count)
        mock_exc.assert_has_calls([fuser_call, fuser_call])




===========Repository Name===========
ironic_lib
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\ironic-lib\ironic_lib\tests\__init__.py
===========File Type===========
.py
===========File Content===========




===========Repository Name===========
networking-ovn
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\networking-ovn\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
envlist = py35,py27,pep8
skipsdist = True

[testenv]
usedevelop = True
install_command = pip install {opts} {packages}
setenv =
   VIRTUAL_ENV={envdir}
   PYTHONWARNINGS=default::DeprecationWarning
deps = -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt}
       -r{toxinidir}/requirements.txt
       -r{toxinidir}/test-requirements.txt
whitelist_externals = bash
                      rm
commands =
  {toxinidir}/tools/ostestr_compat_shim.sh {posargs}
passenv = http_proxy HTTP_PROXY https_proxy HTTPS_PROXY no_proxy NO_PROXY

[testenv:pep8]
basepython = python3
commands = flake8
           {toxinidir}/tools/coding-checks.sh --pylint '{posargs}'
           doc8 doc/source devstack releasenotes/source vagrant rally-jobs
           neutron-db-manage --subproject=networking-ovn check_migration

[testenv:venv]
basepython = python3
commands = {posargs}

[testenv:functional]
setenv =
  {[testenv]setenv}
  OS_TEST_PATH=./networking_ovn/tests/functional
  OS_TEST_TIMEOUT=240
deps = {[testenv]deps}
       -r{toxinidir}/networking_ovn/tests/functional/requirements.txt

[testenv:functional-py35]
basepython = python3.5
setenv =
  {[testenv]setenv}
  OS_TEST_PATH=./networking_ovn/tests/functional
  OS_TEST_TIMEOUT=240
deps = {[testenv]deps}

[testenv:dsvm]
# Fake job to define environment variables shared between dsvm jobs
setenv = OS_TEST_TIMEOUT=240
         OS_LOG_PATH={env:OS_LOG_PATH:/opt/stack/logs}
commands = false

[testenv:dsvm-functional]
setenv = {[testenv:functional]setenv}
         {[testenv:dsvm]setenv}
deps = {[testenv:functional]deps}
commands =
  {toxinidir}/tools/ostestr_compat_shim.sh {posargs}

[testenv:dsvm-functional-py35]
basepython = python3.5
setenv = {[testenv:functional]setenv}
         {[testenv:dsvm]setenv}
deps = {[testenv:functional]deps}
commands =
  {toxinidir}/tools/ostestr_compat_shim.sh {posargs}

[testenv:cover]
basepython = python3
setenv =
  {[testenv]setenv}
  PYTHON=coverage run --source networking_ovn --parallel-mode
commands =
  stestr run --no-subunit-trace {posargs}
  coverage combine
  coverage report --fail-under=70 --skip-covered
  coverage html -d cover
  coverage xml -o cover/coverage.xml

[testenv:docs]
basepython = python3
commands =
  rm -rf doc/build
  doc8 doc/source devstack releasenotes/source vagrant rally-jobs
  sphinx-build -W -b html doc/source doc/build/html

[testenv:debug]
commands = oslo_debug_helper -t networking_ovn/tests {posargs}

[testenv:genconfig]
commands =
    mkdir -p etc/neutron/plugins/ml2
    oslo-config-generator --config-file etc/oslo-config-generator/ml2_conf.ini
    oslo-config-generator --config-file etc/oslo-config-generator/networking_ovn_metadata_agent.ini
whitelist_externals = mkdir

[testenv:releasenotes]
basepython = python3
commands = sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html

[doc8]
# File extensions to check
extensions = .rst

[flake8]
# E123, E125 skipped as they are invalid PEP-8.
# TODO(dougwig) -- uncomment this to test for remaining linkages
# N530 direct neutron imports not allowed

show-source = True
ignore = E123,E125,N530
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build,.tmp
import-order-style = pep8

[hacking]
import_exceptions = networking_ovn
local-check-factory = neutron_lib.hacking.checks.factory

[testenv:lower-constraints]
basepython = python3
deps =
  -c{toxinidir}/lower-constraints.txt
  -r{toxinidir}/test-requirements.txt
  -r{toxinidir}/requirements.txt




===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
skipsdist = True
envlist = linters,docs,releasenotes,inventory,py3-inventory


[testenv]
usedevelop = True
basepython = python2.7
install_command =
    pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
deps =
    -r{toxinidir}/global-requirement-pins.txt
    -r{toxinidir}/test-requirements.txt
passenv =
    COMMON_TESTS_PATH
    HOME
    http_proxy
    HTTP_PROXY
    https_proxy
    HTTPS_PROXY
    no_proxy
    NO_PROXY
    TESTING_BRANCH
    TESTING_HOME
    USER
whitelist_externals =
    bash
setenv =
    PYTHONUNBUFFERED=1
    PYTHONWARNINGS=default::DeprecationWarning
    VIRTUAL_ENV={envdir}
    WORKING_DIR={toxinidir}
    ANSIBLE_EXTRA_ROLE_DIRS={toxinidir}/playbooks/roles:{homedir}/.ansible/roles/ceph-ansible/roles
    ANSIBLE_ROLE_REQUIREMENTS_PATH={toxinidir}/ansible-role-requirements.yml
    TEST_PLAYBOOK={toxinidir}/tests/bootstrap-aio.yml {toxinidir}/playbooks/setup-everything.yml
    ANSIBLE_LINT_PARAMS=--exclude={homedir}/.ansible/roles



[testenv:docs]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands=
    doc8 doc
    bashate tools/build-docs.sh
    {toxinidir}/tools/build-docs.sh



[testenv:deploy-guide]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands = sphinx-build -a -E -W -d deploy-guide/build/doctrees -b html deploy-guide/source deploy-guide/build/html



[doc8]
# Settings for doc8:
extensions = .rst



[testenv:releasenotes]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands =
    sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html



# environment used by the -infra templated docs job
[testenv:venv]
basepython = python3
commands =
    {posargs}



[testenv:pep8]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-pep8.sh"



[flake8]
# Ignores the following rules due to how ansible modules work in general
#     F403 'from ansible.module_utils.basic import *' used;
#           unable to detect undefined names
ignore=F403



[testenv:bashate]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-bashate.sh"



# The deps URL should be set to the appropriate git URL.
# In the tests repo itself, the variable is uniquely set to
# the toxinidir so that the role is able to test itself, but
# the tox config is exactly the same as other repositories.
#
# The value for other repositories must be:
# http://git.openstack.org/cgit/openstack/openstack-ansible-tests/plain/test-ansible-deps.txt
# or for a stable branch:
# http://git.openstack.org/cgit/openstack/openstack-ansible-tests/plain/test-ansible-deps.txt?h=stable/newton
[testenv:ansible]
basepython = python3
deps =
    {[testenv]deps}
    -r{toxinidir}/global-requirement-pins.txt
    -rhttps://git.openstack.org/cgit/openstack/openstack-ansible-tests/plain/test-ansible-deps.txt



[testenv:ansible-syntax]
basepython = python3
deps =
    {[testenv:ansible]deps}
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-syntax.sh"



[testenv:ansible-lint]
basepython = python3
deps =
    {[testenv:ansible]deps}
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-lint.sh"



[testenv:inventory]
basepython = python3
# Use a fixed seed since some inventory tests rely on specific ordering
setenv =
    {[testenv]setenv}
    PYTHONHASHSEED = 100
commands =
    coverage erase
    coverage run -a {toxinidir}/tests/test_inventory.py
    coverage run -a {toxinidir}/tests/test_manage.py
    coverage run -a {toxinidir}/tests/test_dictutils.py
    coverage run -a {toxinidir}/tests/test_ip.py
    coverage run -a {toxinidir}/tests/test_filesystem.py
    coverage report --show-missing --include={toxinidir}/inventory/*,{toxinidir}/osa_toolkit/*



[testenv:py3-inventory]
basepython = python3.5
setenv =
    {[testenv:inventory]setenv}
commands =
    {[testenv:inventory]commands}



[testenv:linters]
basepython = python3
deps =
    {[testenv:docs]deps}
    {[testenv:ansible]deps}
commands =
    {[testenv:pep8]commands}
    {[testenv:bashate]commands}
    {[testenv:ansible-lint]commands}
    {[testenv:ansible-syntax]commands}
    {[testenv:inventory]commands}
    {[testenv:docs]commands}




===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\bootstrap-aio.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2015, Rackspace US, Inc.
#
# 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.

- name: Bootstrap the All-In-One (AIO)
  hosts: localhost
  gather_facts: True
  user: root
  roles:
    - role: "sshd"
    - role: "bootstrap-host"
  vars_files:
    - "{{ playbook_dir }}/../playbooks/defaults/repo_packages/openstack_services.yml"
    - vars/bootstrap-aio-vars.yml
  environment: "{{ deployment_environment_variables | default({}) }}"
  vars:
    ansible_python_interpreter: "/usr/bin/python"
    pip_install_upper_constraints_proto: "{{ ansible_python_version | version_compare('2.7.9', '>=') | ternary('https','http') }}"
    pip_install_upper_constraints: >-
        {{ (playbook_dir ~ '/../global-requirement-pins.txt') | realpath }}
        --constraint {{ pip_install_upper_constraints_proto }}://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt?id={{ requirements_git_install_branch | regex_replace(' #.*$','') }}
    sftp_subsystem:
        'apt': 'sftp /usr/lib/openssh/sftp-server'
        'yum': 'sftp /usr/libexec/openssh/sftp-server'
        'zypper': 'sftp /usr/lib/ssh/sftp-server'
    sshd:
      ListenAddress:
        - 0.0.0.0
        - '::'
      Port: 22
      Protocol: 2
      HostKey:
        - "/etc/ssh/ssh_host_rsa_key"
        - "/etc/ssh/ssh_host_ecdsa_key"
        - "/etc/ssh/ssh_host_ed25519_key"
      UsePrivilegeSeparation: yes
      KeyRegenerationInterval: 3600
      ServerKeyBits: 1024
      SyslogFacility: "AUTH"
      LogLevel: "INFO"
      LoginGraceTime: 120
      StrictModes: yes
      RSAAuthentication: yes
      PubkeyAuthentication: yes
      IgnoreRhosts: yes
      RhostsRSAAuthentication: no
      HostbasedAuthentication: no
      PermitEmptyPasswords: no
      PermitRootLogin: yes
      ChallengeResponseAuthentication: no
      PasswordAuthentication: no
      X11DisplayOffset: 10
      PrintMotd: no
      PrintLastLog: no
      TCPKeepAlive: yes
      AcceptEnv: "LANG LC_*"
      Subsystem: "{{ sftp_subsystem[ansible_pkg_mgr] }}"
      UsePAM: yes
      UseDNS: no
      X11Forwarding: no
      Compression: yes
      CompressionLevel: 6
      MaxSessions: 100
      MaxStartups: "100:100:100"
      GSSAPIAuthentication: no
      GSSAPICleanupCredentials: no
  post_tasks:
    - name: Check that new network interfaces are up
      assert:
        that:
          - ansible_eth12['active'] == true
          - ansible_eth13['active'] == true
          - ansible_eth14['active'] == true
      when:
        - (bootstrap_host_container_tech | default('unknown')) != 'nspawn'

------------->Assertion Roulette .... Multiple things tested in one assert ..... (ansible_eth12)
-------------> Dependence on external config data (/../playbooks/defaults/repo_packages/openstack_services.yml)

===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\test-inventory.ini
===========File Type===========
.ini
===========File Content===========
localhost ansible_connection=local



===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\test-vars-overrides.yml
===========File Type===========
.yml
===========File Content===========
---
# Copyright 2017, Rackspace US, Inc.
#
# 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.

- name: Ensuring group vars
  hosts: "hosts"
  gather_facts: no
  connection: local
  user: root
  tasks:
    - name: Ensuring existing group vars are well applied
      assert:
        that:
          - "openstack_release is defined"
    - name: Ensuring babar is well defined
      assert:
        that:
          - "babar == 'elephant'"
    - name: Ensuring lxc_hosts_package_state is well overridden
      assert:
        that:
          - "lxc_hosts_package_state == 'present'"

- name: Ensuring host vars
  hosts: localhost
  gather_facts: no
  connection: local
  user: root
  tasks:
    - name: Ensuring tintin has milou
      assert:
        that:
          - "tintin == 'milou'"
    - name: Ensuring security_package_state is overridden
      assert:
        that:
          - "security_package_state == 'present'"


---------> Not collecting facts from host , gather_facts: no


===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\test_dictutils.py
===========File Type===========
.py
===========File Content===========
#!/usr/bin/env python

# 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.

import unittest

from osa_toolkit import dictutils as du


class TestMergeDictUnit(unittest.TestCase):
    def test_merging_dict(self):
        base = {'key1': 'value1'}
        target = {'key2': 'value2'}

        new = du.merge_dict(base, target)

        self.assertIn('key2', new.keys())
        self.assertEqual('value2', new['key2'])

    def test_base_dict_is_modified(self):
        base = {'key1': 'value1'}
        target = {'key2': 'value2'}

        new = du.merge_dict(base, target)

        self.assertIs(base, new)

    def test_merging_nested_dicts(self):
        base = {'key1': 'value1'}
        target = {'key2': {'key2.1': 'value2'}}

        new = du.merge_dict(base, target)

        self.assertIn('key2', new.keys())
        self.assertIn('key2.1', new['key2'].keys())

    def test_merging_nested_dicts_with_same_key(self):
        base = {'same_key': {'inside_key1': 'inside_key1'}}
        target = {'same_key': {'inside_key2': 'inside_key2'}}

        new = du.merge_dict(base, target)

        self.assertIn('inside_key1', new['same_key'].keys())
        self.assertIn('inside_key2', new['same_key'].keys())


class TestAppendIfUnit(unittest.TestCase):
    def test_appending_not_present(self):
        base = ['foo', 'bar']
        target = 'baz'

        retval = du.append_if(base, target)

        self.assertIn(target, base)
        self.assertTrue(retval)

    def test_appending_present(self):
        base = ['foo', 'bar', 'baz']
        target = 'baz'

        retval = du.append_if(base, target)

        self.assertFalse(retval)


class TestListRemovalUnit(unittest.TestCase):
    def setUp(self):
        # Can't just use a member list, since it remains changed after each
        # test
        self.base = ['foo', 'bar']

    def test_removing_one_target(self):
        target = ['bar']

        du.recursive_list_removal(self.base, target)

        self.assertNotIn('bar', self.base)

    def test_removing_entire_list(self):
        # Use a copy so we're not hitting the exact same object in memory.
        target = list(self.base)

        du.recursive_list_removal(self.base, target)

        self.assertEqual(0, len(self.base))

    def test_using_base_as_target(self):
        target = self.base

        du.recursive_list_removal(self.base, target)

        self.assertEqual(1, len(self.base))
        self.assertEqual(1, len(target))
        self.assertIn('bar', self.base)

    def test_using_bare_string(self):
        target = 'foo'

        du.recursive_list_removal(self.base, target)

        self.assertEqual(2, len(self.base))


class TestDictRemovalUnit(unittest.TestCase):
    def test_deleting_single_item_in_single_level_noop(self):
        """The function only operates on nested dictionaries"""
        base = {'key1': 'value1'}
        target = ['value1']

        du.recursive_dict_removal(base, target)

        self.assertEqual('value1', base['key1'])

    def test_deleting_single_item(self):
        base = {'key1': {'key1.1': 'value1'}}
        target = ['value1']

        du.recursive_dict_removal(base, target)

        self.assertEqual('value1', base['key1']['key1.1'])

    def test_deleting_single_item_from_list(self):
        base = {'key1': {'key1.1': ['value1']}}
        target = ['value1']

        du.recursive_dict_removal(base, target)

        self.assertEqual(0, len(base['key1']['key1.1']))
        self.assertNotIn('value1', base['key1']['key1.1'])

    def test_deleting_single_item_from_nested_list(self):
        """The function only operates on the 2nd level dictionary"""
        base = {'key1': {'key1.1': {'key1.1.1': ['value1']}}}
        target = ['value1']

        du.recursive_dict_removal(base, target)

        self.assertEqual(1, len(base['key1']['key1.1']['key1.1.1']))
        self.assertIn('value1', base['key1']['key1.1']['key1.1.1'])

    def test_deleting_single_item_top_level_list(self):
        base = {'key1': ['value1']}
        target = ['value1']

        du.recursive_dict_removal(base, target)

        self.assertEqual(0, len(base['key1']))

    def test_deleting_single_nested_key(self):
        base = {'key1': {'key1.1': {'key1.1.1': ['value1']}}}
        target = ['key1.1.1']

        du.recursive_dict_removal(base, target)

        self.assertNotIn('key1.1.1', base['key1']['key1.1'])


if __name__ == '__main__':
    unittest.main()

--------------> Assertion roulette .... Multiple asserts in one test case (test function) .... 


===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\test_filesystem.py
===========File Type===========
.py
===========File Content===========
#!/usr/bin/env python
#
# 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.
#

import mock
import os
from os import path
from osa_toolkit import filesystem as fs
import sys
import unittest

from test_inventory import cleanup
from test_inventory import get_inventory
from test_inventory import make_config

INV_DIR = 'inventory'

sys.path.append(path.join(os.getcwd(), INV_DIR))

TARGET_DIR = path.join(os.getcwd(), 'tests', 'inventory')
USER_CONFIG_FILE = path.join(TARGET_DIR, 'openstack_user_config.yml')


def setUpModule():
    # The setUpModule function is used by the unittest framework.
    make_config()


def tearDownModule():
    # This file should only be removed after all tests are run,
    # thus it is excluded from cleanup.
    os.remove(USER_CONFIG_FILE)


class TestMultipleRuns(unittest.TestCase):
    def test_creating_backup_file(self):
        inventory_file_path = os.path.join(TARGET_DIR,
                                           'openstack_inventory.json')
        get_backup_name_path = 'osa_toolkit.filesystem._get_backup_name'
        backup_name = 'openstack_inventory.json-20160531_171804.json'

        tar_file = mock.MagicMock()
        tar_file.__enter__.return_value = tar_file

        # run make backup with faked tarfiles and date
        with mock.patch('osa_toolkit.filesystem.tarfile.open') as tar_open:
            tar_open.return_value = tar_file
            with mock.patch(get_backup_name_path) as backup_mock:
                backup_mock.return_value = backup_name
                fs._make_backup(TARGET_DIR, inventory_file_path)

        backup_path = path.join(TARGET_DIR, 'backup_openstack_inventory.tar')

        tar_open.assert_called_with(backup_path, 'a')

        # This chain is present because of how tarfile.open is called to
        # make a context manager inside the make_backup function.

        tar_file.add.assert_called_with(inventory_file_path,
                                        arcname=backup_name)

    def test_recreating_files(self):
        # Deleting the files after the first run should cause the files to be
        # completely remade
        get_inventory()

        get_inventory()

        backup_path = path.join(TARGET_DIR, 'backup_openstack_inventory.tar')

        self.assertFalse(os.path.exists(backup_path))

    def test_rereading_files(self):
        # Generate the initial inventory files
        get_inventory(clean=False)

        inv, path = fs.load_inventory(TARGET_DIR)
        self.assertIsInstance(inv, dict)
        self.assertIn('_meta', inv)
        # This test is basically just making sure we get more than
        # INVENTORY_SKEL populated, so we're not going to do deep testing
        self.assertIn('log_hosts', inv)

    def tearDown(self):
        # Clean up here since get_inventory will not do it by design in
        # this test.
        cleanup()


if __name__ == '__main__':
    unittest.main(catchbreak=True)


--------------> Assertion roulette .... Multiple asserts in one test case (test function e.g., test_rereading_files()) .... 
--------------> Dependence on external config data (from test_inventory import make_config)

===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\test_inventory.py
===========File Type===========
.py
===========File Content===========
# 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.

import collections
import copy
import json
import mock
import os
from os import path
try:
    import Queue
except ImportError:
    import queue as Queue
import sys
import unittest
import warnings
import yaml

INV_DIR = 'inventory'

sys.path.append(path.join(os.getcwd(), INV_DIR))

from osa_toolkit import dictutils
import dynamic_inventory
from osa_toolkit import filesystem as fs
from osa_toolkit import generate as di
from osa_toolkit import tools

TARGET_DIR = path.join(os.getcwd(), 'tests', 'inventory')
BASE_ENV_DIR = INV_DIR
CONFIGS_DIR = path.join(os.getcwd(), 'etc', 'openstack_deploy')
CONFD = os.path.join(CONFIGS_DIR, 'conf.d')
AIO_CONFIG_FILE = path.join(CONFIGS_DIR, 'openstack_user_config.yml.aio')
USER_CONFIG_FILE = path.join(TARGET_DIR, 'openstack_user_config.yml')

# These files will be placed in TARGET_DIR by the inventory functions
# They should be cleaned up between each test.
CLEANUP = [
    'openstack_inventory.json',
    'openstack_hostnames_ips.yml',
    'backup_openstack_inventory.tar'
]

# Base config is a global configuration accessible for convenience.
# It should *not* be mutated outside of setUpModule, which populates it.
_BASE_CONFIG = {}


def get_config():
    """Return a copy of the original config so original isn't modified."""
    global _BASE_CONFIG
    return copy.deepcopy(_BASE_CONFIG)


def make_config():
    """Build an inventory configuration from the sample AIO files.

    Take any files specified as '.aio' and load their keys into a
    configuration dict  and write them out to a file for consumption by
    the tests.
    """
    # Allow access here so we can populate the dictionary.
    global _BASE_CONFIG

    _BASE_CONFIG = tools.make_example_config(AIO_CONFIG_FILE, CONFD)

    tools.write_example_config(USER_CONFIG_FILE, _BASE_CONFIG)


def setUpModule():
    # The setUpModule function is used by the unittest framework.
    make_config()


def tearDownModule():
    # This file should only be removed after all tests are run,
    # thus it is excluded from cleanup.
    os.remove(USER_CONFIG_FILE)


def cleanup():
    for f_name in CLEANUP:
        f_file = path.join(TARGET_DIR, f_name)
        if os.path.exists(f_file):
            os.remove(f_file)


def get_inventory(clean=True, extra_args=None):
    "Return the inventory mapping in a dict."
    # Use the list argument to more closely mirror
    # Ansible's use of the callable.
    args = {'config': TARGET_DIR, 'list': True,
            'environment': BASE_ENV_DIR}
    if extra_args:
        args.update(extra_args)
    try:
        inventory_string = di.main(**args)
        inventory = json.loads(inventory_string)
        return inventory
    finally:
        if clean:
            # Remove the file system artifacts since we want to force
            # fresh runs
            cleanup()


class TestArgParser(unittest.TestCase):
    def test_no_args(self):
        arg_dict = dynamic_inventory.args([])
        self.assertIsNone(arg_dict['config'])
        self.assertEqual(arg_dict['list'], False)

    def test_list_arg(self):
        arg_dict = dynamic_inventory.args(['--list'])
        self.assertEqual(arg_dict['list'], True)

    def test_config_arg(self):
        arg_dict = dynamic_inventory.args(['--config',
                                           '/etc/openstack_deploy'])
        self.assertEqual(arg_dict['config'], '/etc/openstack_deploy')


class TestAnsibleInventoryFormatConstraints(unittest.TestCase):
    inventory = None

    expected_groups = [
        'aio1-host_containers',
        'all',
        'all_containers',
        'aodh_alarm_evaluator',
        'aodh_alarm_notifier',
        'aodh_all',
        'aodh_api',
        'aodh_container',
        'aodh_listener',
        'barbican_all',
        'barbican_api',
        'barbican_container',
        'blazar_all',
        'blazar_container',
        'blazar_api',
        'blazar_manager',
        'ceilometer_all',
        'ceilometer_agent_central',
        'ceilometer_agent_compute',
        'ceilometer_agent_notification',
        'ceilometer_central_container',
        'ceph_all',
        'ceph-mon_all',
        'ceph-mon_containers',
        'ceph-mon_container',
        'ceph-mon_hosts',
        'ceph-mon',
        'ceph-osd_all',
        'ceph-osd_containers',
        'ceph-osd_container',
        'ceph-osd_hosts',
        'ceph-osd',
        'ceph-rgw_all',
        'ceph-rgw_containers',
        'ceph-rgw_container',
        'ceph-rgw_hosts',
        'ceph-rgw',
        'cinder_all',
        'cinder_api',
        'cinder_api_container',
        'cinder_backup',
        'cinder_scheduler',
        'cinder_volume',
        'cinder_volumes_container',
        'compute-infra_all',
        'compute-infra_containers',
        'compute-infra_hosts',
        'compute_all',
        'compute_containers',
        'compute_hosts',
        'congress_all',
        'congress_container',
        'congress_server',
        'dashboard_all',
        'dashboard_containers',
        'dashboard_hosts',
        'database_containers',
        'database_hosts',
        'dnsaas_all',
        'dnsaas_containers',
        'dnsaas_hosts',
        'designate_all',
        'designate_container',
        'designate_api',
        'designate_central',
        'designate_mdns',
        'designate_worker',
        'designate_producer',
        'designate_sink',
        'etcd',
        'etcd_all',
        'etcd_container',
        'etcd_containers',
        'etcd_hosts',
        'galera',
        'galera_all',
        'galera_container',
        'glance_all',
        'glance_api',
        'glance_container',
        'glance_registry',
        'gnocchi_all',
        'gnocchi_api',
        'gnocchi_container',
        'gnocchi_metricd',
        'haproxy',
        'haproxy_all',
        'haproxy_container',
        'haproxy_containers',
        'haproxy_hosts',
        'heat_all',
        'heat_api',
        'heat_api_cfn',
        'heat_api_container',
        'heat_engine',
        'horizon',
        'horizon_all',
        'horizon_container',
        'hosts',
        'identity_all',
        'identity_containers',
        'identity_hosts',
        'image_all',
        'image_containers',
        'image_hosts',
        'ironic-infra_all',
        'ironic-infra_containers',
        'ironic-infra_hosts',
        'ironic-server_containers',
        'ironic-server_hosts',
        'ironic_all',
        'ironic_api',
        'ironic_api_container',
        'ironic_conductor',
        'ironic_server',
        'ironic_server_container',
        'ironic_servers',
        'ironic_compute',
        'ironic_compute_container',
        'ironic-compute_containers',
        'ironic-compute_all',
        'ironic-compute_hosts',
        'key-manager_containers',
        'key-manager_hosts',
        'key-manager_all',
        'keystone',
        'keystone_all',
        'keystone_container',
        'kvm-compute_containers',
        'kvm-compute_hosts',
        'log_all',
        'log_containers',
        'log_hosts',
        'lxc_hosts',
        'lxd-compute_containers',
        'lxd-compute_hosts',
        'magnum',
        'magnum-infra_all',
        'magnum-infra_containers',
        'magnum-infra_hosts',
        'magnum_all',
        'magnum_container',
        'mano_all',
        'mano_containers',
        'mano_hosts',
        'nspawn_hosts',
        'octavia-infra_hosts',
        'octavia_all',
        'octavia-api',
        'octavia_server_container',
        'octavia-worker',
        'octavia-housekeeping',
        'octavia-health-manager',
        'octavia-infra_containers',
        'octavia-infra_all',
        'policy_all',
        'policy_containers',
        'policy_hosts',
        'powervm-compute_containers',
        'powervm-compute_hosts',
        'qemu-compute_containers',
        'qemu-compute_hosts',
        'reservation_all',
        'reservation_containers',
        'reservation_hosts',
        'trove_all',
        'trove_api',
        'trove_conductor',
        'trove_taskmanager',
        'trove_api_container',
        'trove-infra_containers',
        'trove-infra_hosts',
        'trove-infra_all',
        'memcached',
        'memcached_all',
        'memcached_container',
        'memcaching_containers',
        'memcaching_hosts',
        'metering-alarm_all',
        'metering-alarm_containers',
        'metering-alarm_hosts',
        'metering-compute_all',
        'metering-compute_container',
        'metering-compute_containers',
        'metering-compute_hosts',
        'metering-infra_all',
        'metering-infra_containers',
        'metering-infra_hosts',
        'metrics_all',
        'metrics_containers',
        'metrics_hosts',
        'mq_containers',
        'mq_hosts',
        'network_all',
        'network_containers',
        'network_hosts',
        'neutron_agent',
        'neutron_agents_container',
        'neutron_all',
        'neutron_bgp_dragent',
        'neutron_dhcp_agent',
        'neutron_l3_agent',
        'neutron_lbaas_agent',
        'neutron_linuxbridge_agent',
        'neutron_metadata_agent',
        'neutron_metering_agent',
        'neutron_openvswitch_agent',
        'neutron_sriov_nic_agent',
        'neutron_server',
        'neutron_server_container',
        'nova_all',
        'nova_api_metadata',
        'nova_api_os_compute',
        'nova_api_container',
        'nova_api_placement',
        'nova_compute',
        'nova_compute_container',
        'nova_conductor',
        'nova_console',
        'nova_scheduler',
        'opendaylight',
        'operator_containers',
        'operator_hosts',
        'orchestration_all',
        'orchestration_containers',
        'orchestration_hosts',
        'os-infra_containers',
        'os-infra_hosts',
        'pkg_repo',
        'rabbit_mq_container',
        'rabbitmq',
        'rabbitmq_all',
        'remote',
        'remote_containers',
        'repo-infra_all',
        'repo-infra_containers',
        'repo-infra_hosts',
        'repo_all',
        'repo_container',
        'rsyslog',
        'rsyslog_all',
        'rsyslog_container',
        'sahara-infra_all',
        'sahara-infra_containers',
        'sahara-infra_hosts',
        'sahara_all',
        'sahara_api',
        'sahara_container',
        'sahara_engine',
        'shared-infra_all',
        'shared-infra_containers',
        'shared-infra_hosts',
        'storage-infra_all',
        'storage-infra_containers',
        'storage-infra_hosts',
        'storage_all',
        'storage_containers',
        'storage_hosts',
        'swift-proxy_all',
        'swift-proxy_containers',
        'swift-proxy_hosts',
        'swift-remote_containers',
        'swift-remote_hosts',
        'swift_acc',
        'swift_acc_container',
        'swift_all',
        'swift_cont',
        'swift_cont_container',
        'swift_containers',
        'swift_hosts',
        'swift_obj',
        'swift_obj_container',
        'swift_proxy',
        'swift_proxy_container',
        'swift_remote',
        'swift_remote_all',
        'swift_remote_container',
        'tacker_all',
        'tacker_container',
        'tacker_server',
        'unbound',
        'unbound_all',
        'unbound_container',
        'unbound_containers',
        'unbound_hosts',
        'utility',
        'utility_all',
        'utility_container'
    ]

    @classmethod
    def setUpClass(cls):
        cls.inventory = get_inventory()

    def test_meta(self):
        meta = self.inventory['_meta']
        self.assertIsNotNone(meta, "_meta missing from inventory")
        self.assertIsInstance(meta, dict, "_meta is not a dict")

    def test_hostvars(self):
        hostvars = self.inventory['_meta']['hostvars']
        self.assertIsNotNone(hostvars, "hostvars missing from _meta")
        self.assertIsInstance(hostvars, dict, "hostvars is not a dict")

    def test_group_vars_all(self):
        group_vars_all = self.inventory['all']
        self.assertIsNotNone(group_vars_all,
                             "group vars all missing from inventory")
        self.assertIsInstance(group_vars_all, dict,
                              "group vars all is not a dict")

        the_vars = group_vars_all['vars']
        self.assertIsNotNone(the_vars,
                             "vars missing from group vars all")
        self.assertIsInstance(the_vars, dict,
                              "vars in group vars all is not a dict")

    def test_expected_host_groups_present(self):

        for group in self.expected_groups:
            the_group = self.inventory[group]
            self.assertIsNotNone(the_group,
                                 "Required host group: %s is missing "
                                 "from inventory" % group)
            self.assertIsInstance(the_group, dict)

            if group != 'all':
                self.assertIn('hosts', the_group)
                self.assertIsInstance(the_group['hosts'], list)

    def test_only_expected_host_groups_present(self):
        all_keys = list(self.expected_groups)
        all_keys.append('_meta')
        self.assertEqual(set(all_keys), set(self.inventory.keys()))

    def test_configured_groups_have_hosts(self):
        config = get_config()
        groups = self.inventory.keys()
        for group in groups:
            if group in config.keys():
                self.assertTrue(0 < len(self.inventory[group]['hosts']))


class TestUserConfiguration(unittest.TestCase):
    def setUp(self):
        self.longMessage = True
        self.loaded_user_configuration = fs.load_user_configuration(TARGET_DIR)

    def test_loading_user_configuration(self):
        """Test that the user configuration can be loaded"""
        self.assertIsInstance(self.loaded_user_configuration, dict)


class TestEnvironments(unittest.TestCase):
    def setUp(self):
        self.longMessage = True
        self.loaded_environment = fs.load_environment(BASE_ENV_DIR, {})

    def test_loading_environment(self):
        """Test that the environment can be loaded"""
        self.assertIsInstance(self.loaded_environment, dict)

    def test_envd_read(self):
        """Test that the env.d contents are inserted into the environment"""
        expected_keys = [
            'component_skel',
            'container_skel',
            'physical_skel',
        ]
        for key in expected_keys:
            self.assertIn(key, self.loaded_environment)


class TestIps(unittest.TestCase):
    def setUp(self):
        # Allow custom assertion errors.
        self.longMessage = True
        self.env = fs.load_environment(BASE_ENV_DIR, {})

    @mock.patch('osa_toolkit.filesystem.load_environment')
    @mock.patch('osa_toolkit.filesystem.load_user_configuration')
    def test_duplicates(self, mock_load_config, mock_load_env):
        """Test that no duplicate IPs are made on any network."""

        # Grab our values read from the file system just once.
        mock_load_config.return_value = get_config()
        mock_load_env.return_value = self.env

        mock_open = mock.mock_open()

        for i in range(0, 99):
            # tearDown is ineffective for this loop, so clean the USED_IPs
            # on each run
            inventory = None
            di.ip.USED_IPS = set()

            # Mock out the context manager being used to write files.
            # We don't need to hit the file system for this test.
            with mock.patch('__main__.open', mock_open):
                inventory = get_inventory()
            ips = collections.defaultdict(int)
            hostvars = inventory['_meta']['hostvars']

            for host, var_dict in hostvars.items():
                nets = var_dict['container_networks']
                for net, vals in nets.items():
                    if 'address' in vals.keys():

                        addr = vals['address']
                        ips[addr] += 1

                        self.assertEqual(1, ips[addr],
                                         msg="IP %s duplicated." % addr)

    def test_empty_ip_queue(self):
        q = Queue.Queue()
        with self.assertRaises(SystemExit) as context:
            # TODO(nrb): import and use ip module directly
            di.ip.get_ip_address('test', q)
        expectedLog = ("Cannot retrieve requested amount of IP addresses. "
                       "Increase the test range in your "
                       "openstack_user_config.yml.")
        self.assertEqual(str(context.exception), expectedLog)

    def tearDown(self):
        # Since the get_ip_address function touches USED_IPS,
        # and USED_IPS is currently a global var, make sure we clean it out
        di.ip.USED_IPS = set()


class TestConfigCheckBase(unittest.TestCase):
    def setUp(self):
        self.config_changed = False
        self.user_defined_config = get_config()

    def delete_config_key(self, user_defined_config, key):
        try:
            if key in user_defined_config:
                del user_defined_config[key]
            elif key in user_defined_config['global_overrides']:
                del user_defined_config['global_overrides'][key]
            else:
                raise KeyError("can't find specified key in user config")
        finally:
            self.write_config()

    def add_config_key(self, key, value):
        self.user_defined_config[key] = value
        self.write_config()

    def add_provider_network(self, net_name, cidr):
        self.user_defined_config['cidr_networks'][net_name] = cidr
        self.write_config()

    def delete_provider_network(self, net_name):
        del self.user_defined_config['cidr_networks'][net_name]
        self.write_config()

    def add_provider_network_key(self, net_name, key, value):
        pns = self.user_defined_config['global_overrides']['provider_networks']
        for net in pns:
            if 'ip_from_q' in net['network']:
                if net['network']['ip_from_q'] == net_name:
                    net['network'][key] = value

    def delete_provider_network_key(self, net_name, key):
        pns = self.user_defined_config['global_overrides']['provider_networks']
        for net in pns:
            if 'ip_from_q' in net['network']:
                if net['network']['ip_from_q'] == net_name:
                    if key in net['network']:
                        del net['network'][key]

    def write_config(self):
        self.config_changed = True
        # Save new user_config_file
        with open(USER_CONFIG_FILE, 'wb') as f:
            f.write(yaml.dump(self.user_defined_config).encode('ascii'))

    def restore_config(self):
        # get back our initial user config file
        self.user_defined_config = get_config()
        self.write_config()

    def set_new_hostname(self, user_defined_config, group,
                         old_hostname, new_hostname):
        # set a new name for the specified hostname
        old_hostname_settings = user_defined_config[group].pop(old_hostname)
        user_defined_config[group][new_hostname] = old_hostname_settings
        self.write_config()

    def set_new_ip(self, user_defined_config, group, hostname, ip):
        # Sets an IP address for a specified host.
        user_defined_config[group][hostname]['ip'] = ip
        self.write_config()

    def add_host(self, group, host_name, ip):
        self.user_defined_config[group][host_name] = {'ip': ip}
        self.write_config()

    def tearDown(self):
        if self.config_changed:
            self.restore_config()


class TestConfigChecks(TestConfigCheckBase):
    def test_missing_container_cidr_network(self):
        self.delete_provider_network('container')
        with self.assertRaises(SystemExit) as context:
            get_inventory()
        expectedLog = ("No container or management network specified in "
                       "user config.")
        self.assertEqual(str(context.exception), expectedLog)

    def test_management_network_malformed(self):
        self.delete_provider_network_key('container', 'is_container_address')
        self.write_config()

        with self.assertRaises(di.ProviderNetworkMisconfiguration) as context:
            get_inventory()
        expectedLog = ("Provider network with queue 'container' "
                       "requires 'is_container_address' "
                       "to be set to True.")
        self.assertEqual(str(context.exception), expectedLog)
        self.restore_config()

    def test_missing_cidr_network_present_in_provider(self):
        self.delete_provider_network('storage')
        with self.assertRaises(SystemExit) as context:
            get_inventory()
        expectedLog = "can't find storage in cidr_networks"
        self.assertEqual(str(context.exception), expectedLog)

    def test_missing_cidr_networks_key(self):
        del self.user_defined_config['cidr_networks']
        self.write_config()
        with self.assertRaises(SystemExit) as context:
            get_inventory()
        expectedLog = "No container CIDR specified in user config"
        self.assertEqual(str(context.exception), expectedLog)

    def test_provider_networks_check(self):
        # create config file without provider networks
        self.delete_config_key(self.user_defined_config, 'provider_networks')
        # check if provider networks absence is Caught
        with self.assertRaises(SystemExit) as context:
            get_inventory()
        expectedLog = "provider networks can't be found under global_overrides"
        self.assertIn(expectedLog, str(context.exception))

    def test_global_overrides_check(self):
        # create config file without global_overrides
        self.delete_config_key(self.user_defined_config, 'global_overrides')
        # check if global_overrides absence is Caught
        with self.assertRaises(SystemExit) as context:
            get_inventory()
        expectedLog = "global_overrides can't be found in user config"
        self.assertEqual(str(context.exception), expectedLog)

    def test_two_hosts_same_ip(self):
        # Use an OrderedDict to be certain our testing order is preserved
        # Even with the same hash seed, different OSes get different results,
        # eg. local OS X vs gate's Linux
        config = collections.OrderedDict()
        config['shared-infra_hosts'] = {
            'host1': {
                'ip': '192.168.1.1'
            }
        }
        config['compute_hosts'] = {
            'host2': {
                'ip': '192.168.1.1'
            }
        }

        with self.assertRaises(di.MultipleHostsWithOneIPError) as context:
            di._check_same_ip_to_multiple_host(config)
        self.assertEqual(context.exception.ip, '192.168.1.1')
        self.assertEqual(context.exception.assigned_host, 'host1')
        self.assertEqual(context.exception.new_host, 'host2')

    def test_two_hosts_same_ip_externally(self):
        self.set_new_hostname(self.user_defined_config, "haproxy_hosts",
                              "aio1", "hap")
        with self.assertRaises(di.MultipleHostsWithOneIPError) as context:
            get_inventory()
        expectedLog = ("Both host:aio1 and host:hap have "
                       "address:172.29.236.100 assigned.  Cannot "
                       "assign same ip to both hosts")
        self.assertEqual(str(context.exception), expectedLog)

    def test_one_host_two_ips_externally(self):
        # haproxy chosen because it was last in the config file as of
        # writing
        self.set_new_ip(self.user_defined_config, 'haproxy_hosts', 'aio1',
                        '172.29.236.101')
        with self.assertRaises(di.MultipleIpForHostError) as context:
            get_inventory()
        expectedLog = ("Host aio1 has both 172.29.236.100 and 172.29.236.101 "
                       "assigned")
        self.assertEqual(str(context.exception), expectedLog)

    def test_two_ips(self):
        # Use an OrderedDict to be certain our testing order is preserved
        # Even with the same hash seed, different OSes get different results,
        # eg. local OS X vs gate's Linux
        config = collections.OrderedDict()
        config['shared-infra_hosts'] = {
            'host1': {
                'ip': '192.168.1.1'
            }
        }
        config['compute_hosts'] = {
            'host1': {
                'ip': '192.168.1.2'
            }
        }

        with self.assertRaises(di.MultipleIpForHostError) as context:
            di._check_multiple_ips_to_host(config)
        self.assertEqual(context.exception.current_ip, '192.168.1.1')
        self.assertEqual(context.exception.new_ip, '192.168.1.2')
        self.assertEqual(context.exception.hostname, 'host1')

    def test_correct_hostname_ip_map(self):
        config = {
            'shared-infra_hosts': {
                'host1': {
                    'ip': '192.168.1.1'
                }
            },
            'compute_hosts': {
                'host2': {
                    'ip': '192.168.1.2'
                }
            },
        }
        ret = di._check_multiple_ips_to_host(config)
        self.assertTrue(ret)


class TestStaticRouteConfig(TestConfigCheckBase):
    def setUp(self):
        super(TestStaticRouteConfig, self).setUp()
        self.expectedMsg = ("Static route provider network with queue "
                            "'container' needs both 'cidr' and 'gateway' "
                            "values.")

    def add_static_route(self, q_name, route_dict):
        """Adds a static route to a provider network."""
        pn = self.user_defined_config['global_overrides']['provider_networks']
        for net in pn:
            net_dict = net['network']
            q = net_dict.get('ip_from_q', None)
            if q == q_name:
                net_dict['static_routes'] = [route_dict]
        self.write_config()

    def test_setting_static_route(self):
        route_dict = {'cidr': '10.176.0.0/12',
                      'gateway': '172.29.248.1'}
        self.add_static_route('container', route_dict)
        inventory = get_inventory()

        # Use aio1 and 'container_address' since they're known keys.
        hostvars = inventory['_meta']['hostvars']['aio1']
        cont_add = hostvars['container_networks']['container_address']

        self.assertIn('static_routes', cont_add)

        first_route = cont_add['static_routes'][0]
        self.assertIn('cidr', first_route)
        self.assertIn('gateway', first_route)

    def test_setting_bad_static_route_only_cidr(self):
        route_dict = {'cidr': '10.176.0.0/12'}
        self.add_static_route('container', route_dict)

        with self.assertRaises(di.MissingStaticRouteInfo) as context:
            get_inventory()

        exception = context.exception

        self.assertEqual(str(exception), self.expectedMsg)

    def test_setting_bad_static_route_only_gateway(self):
        route_dict = {'gateway': '172.29.248.1'}
        self.add_static_route('container', route_dict)

        with self.assertRaises(di.MissingStaticRouteInfo) as context:
            get_inventory()

        exception = context.exception

        self.assertEqual(exception.message, self.expectedMsg)

    def test_setting_bad_gateway_value(self):
        route_dict = {'cidr': '10.176.0.0/12',
                      'gateway': None}
        self.add_static_route('container', route_dict)

        with self.assertRaises(di.MissingStaticRouteInfo) as context:
            get_inventory()

        exception = context.exception

        self.assertEqual(exception.message, self.expectedMsg)

    def test_setting_bad_cidr_value(self):
        route_dict = {'cidr': None,
                      'gateway': '172.29.248.1'}
        self.add_static_route('container', route_dict)

        with self.assertRaises(di.MissingStaticRouteInfo) as context:
            get_inventory()

        exception = context.exception

        self.assertEqual(exception.message, self.expectedMsg)

    def test_setting_bad_cidr_gateway_value(self):
        route_dict = {'cidr': None,
                      'gateway': None}
        self.add_static_route('container', route_dict)

        with self.assertRaises(di.MissingStaticRouteInfo) as context:
            get_inventory()

        exception = context.exception

        self.assertEqual(exception.message, self.expectedMsg)


class TestGlobalOverridesConfigDeletion(TestConfigCheckBase):
    def setUp(self):
        super(TestGlobalOverridesConfigDeletion, self).setUp()
        self.inventory = get_inventory()

    def add_global_override(self, var_name, var_value):
        """Adds an arbitrary name and value to the global_overrides dict."""
        overrides = self.user_defined_config['global_overrides']
        overrides[var_name] = var_value

    def remove_global_override(self, var_name):
        """Removes target key from the global_overrides dict."""
        overrides = self.user_defined_config['global_overrides']
        del overrides[var_name]

    def test_global_overrides_delete_when_merge(self):
        """Vars removed from global overrides are removed from inventory"""
        self.add_global_override('foo', 'bar')

        di._parse_global_variables({}, self.inventory,
                                   self.user_defined_config)

        self.remove_global_override('foo')

        di._parse_global_variables({}, self.inventory,
                                   self.user_defined_config)

        self.assertNotIn('foo', self.inventory['all']['vars'],
                         "foo var not removed from group_vars_all")

    def test_global_overrides_merge(self):
        self.add_global_override('foo', 'bar')

        di._parse_global_variables({}, self.inventory,
                                   self.user_defined_config)

        self.assertEqual('bar', self.inventory['all']['vars']['foo'])

    def test_container_cidr_key_retained(self):
        user_cidr = self.user_defined_config['cidr_networks']['container']
        di._parse_global_variables(user_cidr, self.inventory,
                                   self.user_defined_config)
        self.assertIn('container_cidr', self.inventory['all']['vars'])
        self.assertEqual(self.inventory['all']['vars']['container_cidr'],
                         user_cidr)

    def test_only_old_vars_deleted(self):
        self.inventory['all']['vars']['foo'] = 'bar'

        di._parse_global_variables('', self.inventory,
                                   self.user_defined_config)

        self.assertNotIn('foo', self.inventory['all']['vars'])

    def test_empty_vars(self):
        del self.inventory['all']

        di._parse_global_variables('', self.inventory,
                                   self.user_defined_config)

        self.assertIn('container_cidr', self.inventory['all']['vars'])

        for key in self.user_defined_config['global_overrides']:
            self.assertIn(key, self.inventory['all']['vars'])


class TestEnsureInventoryUptoDate(unittest.TestCase):
    def setUp(self):
        self.env = fs.load_environment(BASE_ENV_DIR, {})
        # Copy because we manipulate the structure in each test;
        # not copying would modify the global var in the target code
        self.inv = copy.deepcopy(di.INVENTORY_SKEL)
        # Since we're not running skel_setup, add necessary keys
        self.host_vars = self.inv['_meta']['hostvars']

        # The _ensure_inventory_uptodate function depends on values inserted
        # by the skel_setup function
        di.skel_setup(self.env, self.inv)

    def test_missing_required_host_vars(self):
        self.host_vars['host1'] = {}

        di._ensure_inventory_uptodate(self.inv, self.env['container_skel'])

        for required_key in di.REQUIRED_HOSTVARS:
            self.assertIn(required_key, self.host_vars['host1'])

    def test_missing_container_name(self):
        self.host_vars['host1'] = {}

        di._ensure_inventory_uptodate(self.inv, self.env['container_skel'])

        self.assertIn('container_name', self.host_vars['host1'])
        self.assertEqual(self.host_vars['host1']['container_name'], 'host1')

    def test_inserting_container_networks_is_dict(self):
        self.host_vars['host1'] = {}

        di._ensure_inventory_uptodate(self.inv, self.env['container_skel'])

        self.assertIsInstance(self.host_vars['host1']['container_networks'],
                              dict)

    def test_populating_inventory_info(self):
        skel = self.env['container_skel']

        di._ensure_inventory_uptodate(self.inv, skel)

        for container_type, type_vars in skel.items():
            hosts = self.inv[container_type]['hosts']
            if hosts:
                for host in hosts:
                    host_var_entries = self.inv['_meta']['hostvars'][host]
                    if 'properties' in type_vars:
                        self.assertEqual(host_var_entries['properties'],
                                         type_vars['properties'])

    def tearDown(self):
        self.env = None
        self.host_vars = None
        self.inv = None


class OverridingEnvBase(unittest.TestCase):
    def setUp(self):
        self.base_env = fs.load_environment(BASE_ENV_DIR, {})

        # Use the cinder configuration as our sample for override testing
        with open(path.join(BASE_ENV_DIR, 'env.d', 'cinder.yml'), 'r') as f:
            self.cinder_config = yaml.safe_load(f.read())

        self.override_path = path.join(TARGET_DIR, 'env.d')
        os.mkdir(self.override_path)

    def write_override_env(self):
        with open(path.join(self.override_path, 'cinder.yml'), 'w') as f:
            f.write(yaml.safe_dump(self.cinder_config))

    def tearDown(self):
        os.remove(path.join(self.override_path, 'cinder.yml'))
        os.rmdir(self.override_path)


class TestOverridingEnvVars(OverridingEnvBase):

    def test_cinder_metal_override(self):
        vol = self.cinder_config['container_skel']['cinder_volumes_container']
        vol['properties']['is_metal'] = False

        self.write_override_env()

        fs.load_environment(TARGET_DIR, self.base_env)

        test_vol = self.base_env['container_skel']['cinder_volumes_container']
        self.assertFalse(test_vol['properties']['is_metal'])

    def test_deleting_elements(self):
        # Leave only the 'properties' dictionary attached to simulate writing
        # a partial override file

        vol = self.cinder_config['container_skel']['cinder_volumes_container']
        keys = vol.keys()
        to_delete = []
        for key in vol.keys():
            if not key == 'properties':
                to_delete.append(key)

        for key in to_delete:
            del vol[key]

        self.write_override_env()

        fs.load_environment(TARGET_DIR, self.base_env)

        test_vol = self.base_env['container_skel']['cinder_volumes_container']

        self.assertIn('belongs_to', test_vol)

    def test_adding_new_keys(self):
        vol = self.cinder_config['container_skel']['cinder_volumes_container']
        vol['a_new_key'] = 'Added'

        self.write_override_env()

        fs.load_environment(TARGET_DIR, self.base_env)

        test_vol = self.base_env['container_skel']['cinder_volumes_container']

        self.assertIn('a_new_key', test_vol)
        self.assertEqual(test_vol['a_new_key'], 'Added')

    def test_emptying_dictionaries(self):
        self.cinder_config['container_skel']['cinder_volumes_container'] = {}

        self.write_override_env()

        fs.load_environment(TARGET_DIR, self.base_env)

        test_vol = self.base_env['container_skel']['cinder_volumes_container']

        self.assertNotIn('belongs_to', test_vol)

    def test_emptying_lists(self):
        vol = self.cinder_config['container_skel']['cinder_volumes_container']
        vol['belongs_to'] = []

        self.write_override_env()

        fs.load_environment(TARGET_DIR, self.base_env)

        test_vol = self.base_env['container_skel']['cinder_volumes_container']

        self.assertEqual(test_vol['belongs_to'], [])


class TestOverridingEnvIntegration(OverridingEnvBase):
    def setUp(self):
        super(TestOverridingEnvIntegration, self).setUp()
        self.user_defined_config = get_config()

        # Inventory is necessary since keys are assumed present
        self.inv, path = fs.load_inventory(TARGET_DIR, di.INVENTORY_SKEL)

    def skel_setup(self):
        self.environment = fs.load_environment(TARGET_DIR, self.base_env)

        di.skel_setup(self.environment, self.inv)

        di.skel_load(
            self.environment.get('physical_skel'),
            self.inv
        )

    def test_emptying_container_integration(self):
        self.cinder_config = {}
        self.cinder_config['container_skel'] = {'cinder_volumes_container': {}}

        self.write_override_env()
        self.skel_setup()

        di.container_skel_load(
            self.environment.get('container_skel'),
            self.inv,
            self.user_defined_config
        )

        test_vol = self.base_env['container_skel']['cinder_volumes_container']

        self.assertNotIn('belongs_to', test_vol)
        self.assertNotIn('contains', test_vol)

    def test_empty_contains(self):
        vol = self.cinder_config['container_skel']['cinder_volumes_container']
        vol['contains'] = []

        self.write_override_env()
        self.skel_setup()

        di.container_skel_load(
            self.environment.get('container_skel'),
            self.inv,
            self.user_defined_config
        )

        test_vol = self.base_env['container_skel']['cinder_volumes_container']

        self.assertEqual(test_vol['contains'], [])

    def test_empty_belongs_to(self):
        vol = self.cinder_config['container_skel']['cinder_volumes_container']
        vol['belongs_to'] = []

        self.write_override_env()
        self.skel_setup()

        di.container_skel_load(
            self.environment.get('container_skel'),
            self.inv,
            self.user_defined_config
        )

        test_vol = self.base_env['container_skel']['cinder_volumes_container']

        self.assertEqual(test_vol['belongs_to'], [])

    def tearDown(self):
        super(TestOverridingEnvIntegration, self).tearDown()
        self.user_defined_config = None
        self.inv = None


class TestSetUsedIPS(unittest.TestCase):
    def setUp(self):
        # Clean up the used ips in case other tests didn't.
        di.ip.USED_IPS = set()

        # Create a fake inventory just for this test.
        self.inventory = {'_meta': {'hostvars': {
            'host1': {'container_networks': {
                'net': {'address': '172.12.1.1'}
            }},
            'host2': {'container_networks': {
                'net': {'address': '172.12.1.2'}
            }},
        }}}

    def test_adding_inventory_used_ips(self):
        config = {'used_ips': None}

        # TODO(nrb): This is a smell, needs to set more directly

        di.ip.set_used_ips(config, self.inventory)

        self.assertEqual(len(di.ip.USED_IPS), 2)
        self.assertIn('172.12.1.1', di.ip.USED_IPS)
        self.assertIn('172.12.1.2', di.ip.USED_IPS)

    def tearDown(self):
        di.ip.USED_IPS = set()


class TestConfigCheckFunctional(TestConfigCheckBase):
    def duplicate_ip(self):
        ip = self.user_defined_config['log_hosts']['aio1']
        self.user_defined_config['log_hosts']['bogus'] = ip

    def test_checking_good_config(self):
        output = di.main(config=TARGET_DIR, check=True,
                         environment=BASE_ENV_DIR)
        self.assertEqual(output, 'Configuration ok!')

    def test_duplicated_ip(self):
        self.duplicate_ip()
        self.write_config()
        with self.assertRaises(di.MultipleHostsWithOneIPError) as context:
            di.main(config=TARGET_DIR, check=True, environment=BASE_ENV_DIR)
        self.assertEqual(context.exception.ip, '172.29.236.100')


class TestNetworkEntry(unittest.TestCase):
    def test_all_args_filled(self):
        entry = di.network_entry(True, 'eth1', 'br-mgmt', 'my_type', '1700')

        self.assertNotIn('interface', entry.keys())
        self.assertEqual(entry['bridge'], 'br-mgmt')
        self.assertEqual(entry['type'], 'my_type')
        self.assertEqual(entry['mtu'], '1700')

    def test_container_dict(self):
        entry = di.network_entry(False, 'eth1', 'br-mgmt', 'my_type', '1700')

        self.assertEqual(entry['interface'], 'eth1')


class TestDebugLogging(unittest.TestCase):
    @mock.patch('osa_toolkit.generate.logging')
    @mock.patch('osa_toolkit.generate.logger')
    def test_logging_enabled(self, mock_logger, mock_logging):
        # Shadow the real value so tests don't complain about it
        mock_logging.DEBUG = 10

        get_inventory(extra_args={"debug": True})

        self.assertTrue(mock_logging.basicConfig.called)
        self.assertTrue(mock_logger.info.called)
        self.assertTrue(mock_logger.debug.called)

    @mock.patch('osa_toolkit.generate.logging')
    @mock.patch('osa_toolkit.generate.logger')
    def test_logging_disabled(self, mock_logger, mock_logging):
        get_inventory(extra_args={"debug": False})

        self.assertFalse(mock_logging.basicConfig.called)
        # Even though logging is disabled, we still call these
        # all over the place; they just choose not to do anything.
        # NOTE: No info messages are published when debug is False
        self.assertTrue(mock_logger.debug.called)


class TestLxcHosts(TestConfigCheckBase):

    def test_lxc_hosts_group_present(self):
        inventory = get_inventory()
        self.assertIn('lxc_hosts', inventory)

    def test_lxc_hosts_only_inserted_once(self):
        inventory = get_inventory()
        self.assertEqual(1, len(inventory['lxc_hosts']['hosts']))

    def test_lxc_hosts_members(self):
        self.add_host('shared-infra_hosts', 'aio2', '172.29.236.101')
        inventory = get_inventory()
        self.assertIn('aio2', inventory['lxc_hosts']['hosts'])
        self.assertIn('aio1', inventory['lxc_hosts']['hosts'])

    def test_lxc_hosts_in_config_raises_error(self):
        self.add_config_key('lxc_hosts', {})
        with self.assertRaises(di.LxcHostsDefined):
            get_inventory()

    def test_host_without_containers(self):
        self.add_host('compute_hosts', 'compute1', '172.29.236.102')
        inventory = get_inventory()
        self.assertNotIn('compute1', inventory['lxc_hosts']['hosts'])

    def test_cleaning_bad_hosts(self):
        self.add_host('compute_hosts', 'compute1', '172.29.236.102')
        inventory = get_inventory()
        # insert compute1 into lxc_hosts, which mimicks bug behavior
        inventory['lxc_hosts']['hosts'].append('compute1')
        faked_path = INV_DIR

        with mock.patch('osa_toolkit.filesystem.load_inventory') as inv_mock:
            inv_mock.return_value = (inventory, faked_path)
            new_inventory = get_inventory()
        # host should no longer be in lxc_hosts

        self.assertNotIn('compute1', new_inventory['lxc_hosts']['hosts'])

    def test_emptying_lxc_hosts(self):
        """If lxc_hosts is deleted between runs, it should re-populate"""

        inventory = get_inventory()
        original_lxc_hosts = inventory.pop('lxc_hosts')

        self.assertNotIn('lxc_hosts', inventory.keys())

        faked_path = INV_DIR
        with mock.patch('osa_toolkit.filesystem.load_inventory') as inv_mock:
            inv_mock.return_value = (inventory, faked_path)
            new_inventory = get_inventory()

        self.assertEqual(original_lxc_hosts, new_inventory['lxc_hosts'])


class TestConfigMatchesEnvironment(unittest.TestCase):
    def setUp(self):
        self.env = fs.load_environment(BASE_ENV_DIR, {})

    def test_matching_keys(self):
        config = get_config()

        result = di._check_all_conf_groups_present(config, self.env)
        self.assertTrue(result)

    def test_failed_match(self):
        bad_config = get_config()
        bad_config['bogus_key'] = []

        result = di._check_all_conf_groups_present(bad_config, self.env)
        self.assertFalse(result)

    def test_extra_config_key_warning(self):
        bad_config = get_config()
        bad_config['bogus_key'] = []
        with warnings.catch_warnings(record=True) as wl:
            di._check_all_conf_groups_present(bad_config, self.env)
            self.assertEqual(1, len(wl))
            self.assertIn('bogus_key', str(wl[0].message))

    def test_multiple_extra_keys(self):
        bad_config = get_config()
        bad_config['bogus_key1'] = []
        bad_config['bogus_key2'] = []

        with warnings.catch_warnings(record=True) as wl:
            di._check_all_conf_groups_present(bad_config, self.env)
            self.assertEqual(2, len(wl))
            warn_msgs = [str(warn.message) for warn in wl]
            warn_msgs.sort()
            self.assertIn('bogus_key1', warn_msgs[0])
            self.assertIn('bogus_key2', warn_msgs[1])

    def test_confirm_exclusions(self):
        """Ensure the excluded keys in the function are present."""
        config = get_config()
        excluded_keys = ('global_overrides', 'cidr_networks', 'used_ips')

        for key in excluded_keys:
            config[key] = 'sentinel value'

        with warnings.catch_warnings(record=True) as wl:
            di._check_all_conf_groups_present(config, self.env)
            self.assertEqual(0, len(wl))

        for key in excluded_keys:
            self.assertIn(key, config.keys())


class TestInventoryGroupConstraints(unittest.TestCase):
    def setUp(self):
        self.env = fs.load_environment(BASE_ENV_DIR, {})

    def test_group_with_hosts_dont_have_children(self):
        """Require that groups have children groups or hosts, not both."""
        inventory = get_inventory()

        # This should only work on groups, but stuff like '_meta' and 'all'
        # are in here, too.
        for key, values in inventory.items():
            # The keys for children/hosts can exist, the important part is being empty lists.
            has_children = bool(inventory.get('children'))
            has_hosts = bool(inventory.get('hosts'))

            self.assertFalse(has_children and has_hosts)

    def _create_bad_env(self, env):
        # This environment setup is used because it was reported with
        # bug #1646136
        override = """
            physical_skel:
              local-compute_containers:
                belongs_to:
                - compute_containers
              local-compute_hosts:
                belongs_to:
                - compute_hosts
              rbd-compute_containers:
                belongs_to:
                - compute_containers
              rbd-compute_hosts:
                belongs_to:
                - compute_hosts
        """

        bad_env = yaml.safe_load(override)

        # This is essentially what load_environment does, after all the file
        # system walking
        dictutils.merge_dict(env, bad_env)

        return env

    def test_group_with_hosts_and_children_fails(self):
        """Integration test making sure the whole script fails."""
        env = self._create_bad_env(self.env)


        config = get_config()

        kwargs = {
            'load_environment': mock.DEFAULT,
            'load_user_configuration': mock.DEFAULT
        }

        with mock.patch.multiple('osa_toolkit.filesystem', **kwargs) as mocks:
            mocks['load_environment'].return_value = env
            mocks['load_user_configuration'].return_value = config

            with self.assertRaises(di.GroupConflict) as context:
                get_inventory()

    def test_group_validation_unit(self):
        env = self._create_bad_env(self.env)

        config = get_config()

        with self.assertRaises(di.GroupConflict):
            di._check_group_branches(config, env['physical_skel'])

    def test_group_validation_no_config(self):
        result = di._check_group_branches(None, self.env)
        self.assertTrue(result)

    def test_group_validation_passes_defaults(self):
        config = get_config()

        result = di._check_group_branches(config, self.env['physical_skel'])

        self.assertTrue(result)

class TestL3ProviderNetworkConfig(TestConfigCheckBase):
    def setUp(self):
        super(TestL3ProviderNetworkConfig, self).setUp()
        self.delete_provider_network('container')
        self.add_provider_network('pod1_container', '172.29.236.0/22')
        self.add_provider_network_key('container', 'ip_from_q',
                                      'pod1_container')
        self.add_provider_network_key('pod1_container', 'address_prefix',
                                      'management')
        self.add_provider_network_key('pod1_container', 'reference_group',
                                      'pod1_hosts')
        self.add_config_key('pod1_hosts', {})
        self.add_host('pod1_hosts', 'aio2', '172.29.236.101')
        self.add_host('compute_hosts', 'aio2', '172.29.236.101')
        self.write_config()
        self.inventory = get_inventory()

    def test_address_prefix_name_applied(self):
         aio2_host_vars = self.inventory['_meta']['hostvars']['aio2']
         aio2_container_networks = aio2_host_vars['container_networks']
         self.assertIsInstance(aio2_container_networks['management_address'],
                               dict)

    def test_host_outside_reference_group_excluded(self):
         aio1_host_vars = self.inventory['_meta']['hostvars']['aio1']
         aio1_container_networks = aio1_host_vars['container_networks']
         self.assertNotIn('management_address', aio1_container_networks)

if __name__ == '__main__':
    unittest.main(catchbreak=True)


-------------------> Assertion roulette .... TestDebugLogging/TestNetworkEntry
--------------> Dependence on external config data (AIO_CONFIG_FILE = path.join(CONFIGS_DIR, 'openstack_user_config.yml.aio')USER_CONFIG_FILE = path.join(TARGET_DIR, 'openstack_user_config.yml'))

===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\test_ip.py
===========File Type===========
.py
===========File Content===========
#!/usr/bin/env python

# 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.

import unittest

from osa_toolkit import ip


class TestIPManager(unittest.TestCase):
    def test_basic_instantiation(self):
        manager = ip.IPManager()

        self.assertEqual({}, manager.queues)
        self.assertEqual(set(), manager.used)

    def test_verbose_instantiation(self):
        manager = ip.IPManager(queues={'test': '192.168.0.0/24'},
                               used_ips=set(['192.168.0.0', '192.168.0.255']))
        self.assertEqual(2, len(manager.used))
        self.assertEqual(254, len(manager.queues['test']))

    def test_instantiation_with_used_list(self):
        manager = ip.IPManager(used_ips=['192.168.0.0', '192.168.0.255'])

        self.assertEqual(2, len(manager.used))

    def test_verbose_instantiation_duplicated_ips(self):
        manager = ip.IPManager(used_ips=['192.168.0.0', '192.168.0.0'])

        self.assertEqual(1, len(manager.used))

    def test_deleting_used(self):
        manager = ip.IPManager(used_ips=set(['192.168.1.1']))

        del manager.used

        self.assertEqual(set(), manager.used)

    def test_getitem(self):
        manager = ip.IPManager(queues={'test': '192.168.0.0/24'})

        self.assertEqual(manager.queues['test'], manager['test'])

    def test_loading_queue(self):
        manager = ip.IPManager()
        manager.load('test', '192.168.0.0/24')
        self.assertEqual(254, len(manager.queues['test']))

    def test_loading_network_excludes(self):
        manager = ip.IPManager()
        manager.load('test', '192.168.0.0/24')
        self.assertNotIn('192.168.0.0', manager.queues['test'])
        self.assertNotIn('192.168.0.255', manager.queues['test'])

    def test_loading_used_ips(self):
        manager = ip.IPManager()
        manager.load('test', '192.168.0.0/24')

        self.assertEqual(2, len(manager.used))
        self.assertIn('192.168.0.0', manager.used)
        self.assertIn('192.168.0.255', manager.used)

    def test_load_creates_networks(self):
        manager = ip.IPManager()
        manager.load('test', '192.168.0.0/24')

        self.assertIn('test', manager._networks)

    def test_loaded_randomly(self):
        manager = ip.IPManager()
        manager.load('test', '192.168.0.0/24')

        self.assertNotEqual(['192.168.0.1', '192.168.0.2', '192.168.0.3'],
                            manager.queues['test'][0:3])

    def test_getting_ip(self):
        manager = ip.IPManager(queues={'test': '192.168.0.0/24'})
        my_ip = manager.get('test')

        self.assertTrue(my_ip.startswith('192.168.0'))
        self.assertIn(my_ip, manager.used)
        self.assertNotIn(my_ip, manager.queues['test'])

    def test_getting_ip_from_empty_queue(self):
        manager = ip.IPManager(queues={'test': '192.168.0.0/31'})
        # There will only be 1 usable IP address in this range.
        manager.get('test')

        with self.assertRaises(ip.EmptyQueue):
            manager.get('test')

    def test_get_ip_from_missing_queue(self):
        manager = ip.IPManager()

        with self.assertRaises(ip.NoSuchQueue):
            manager.get('management')

    def test_release_used_ip(self):
        target_ip = '192.168.0.1'
        manager = ip.IPManager(queues={'test': '192.168.0.0/31'},
                               used_ips=[target_ip])

        manager.release(target_ip)

        # No broadcast address on this network, so only the network addr left
        self.assertEqual(1, len(manager.used))
        self.assertNotIn(target_ip, manager.used)
        self.assertIn(target_ip, manager['test'])

    def test_save_not_implemented(self):
        manager = ip.IPManager()

        with self.assertRaises(NotImplementedError):
            manager.save()

    def test_queue_dict_copied(self):
        manager = ip.IPManager(queues={'test': '192.168.0.0/31'})
        external = manager.queues
        self.assertIsNot(manager.queues, external)
        self.assertIsNot(manager.queues['test'], external['test'])

    def test_queue_list_copied(self):
        manager = ip.IPManager(queues={'test': '192.168.0.0/31'})
        external = manager['test']
        # test against the internal structure since .queues should
        # itself be making copies
        self.assertIsNot(manager._queues['test'], external)

    def test_used_ips_copies(self):
        manager = ip.IPManager(used_ips=['192.168.0.1'])
        external = manager.used
        self.assertIsNot(manager._used_ips, external)

    def test_deleting_used_ips_releases_to_queues(self):
        target_ip = '192.168.0.1'
        manager = ip.IPManager(queues={'test': '192.168.0.0/31'},
                               used_ips=[target_ip])

        del manager.used

        self.assertIn(target_ip, manager['test'])


if __name__ == "__main__":
    unittest.main()


-------------------> Assertion roulette .... test_release_used_ip
-------------------> Need to skip Python files overall 

===========Repository Name===========
openstack-ansible
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible\tests\test_manage.py
===========File Type===========
.py
===========File Content===========
#!/usr/bin/env python

# 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.

import os
from os import path
from osa_toolkit import manage as mi
import test_inventory
import unittest

TARGET_DIR = path.join(os.getcwd(), 'tests', 'inventory')


def setUpModule():
    test_inventory.make_config()


def tearDownModule():
    os.remove(test_inventory.USER_CONFIG_FILE)


class TestExportFunction(unittest.TestCase):
    def setUp(self):
        self.inv = test_inventory.get_inventory()

    def tearDown(self):
        test_inventory.cleanup()

    def test_host_is_present(self):
        host_inv = mi.export_host_info(self.inv)['hosts']
        self.assertIn('aio1', host_inv.keys())

    def test_groups_added(self):
        host_inv = mi.export_host_info(self.inv)['hosts']
        self.assertIn('groups', host_inv['aio1'].keys())

    def test_variables_added(self):
        host_inv = mi.export_host_info(self.inv)['hosts']
        self.assertIn('hostvars', host_inv['aio1'].keys())

    def test_number_of_hosts(self):
        host_inv = mi.export_host_info(self.inv)['hosts']

        self.assertEqual(len(self.inv['_meta']['hostvars']),
                         len(host_inv))

    def test_all_information_added(self):
        all_info = mi.export_host_info(self.inv)['all']
        self.assertIn('provider_networks', all_info)

    def test_all_lb_information(self):
        all_info = mi.export_host_info(self.inv)['all']
        inv_all = self.inv['all']['vars']
        self.assertEqual(inv_all['internal_lb_vip_address'],
                         all_info['internal_lb_vip_address'])


class TestRemoveIpfunction(unittest.TestCase):
    def setUp(self):
        self.inv = test_inventory.get_inventory()

    def tearDown(self):
        test_inventory.cleanup()

    def test_ips_removed(self):
        mi.remove_ip_addresses(self.inv)
        mi.remove_ip_addresses(self.inv, TARGET_DIR)
        hostvars = self.inv['_meta']['hostvars']

        for host, variables in hostvars.items():
            has_networks = 'container_networks' in variables
            if variables.get('is_metal', False):
                continue
            self.assertFalse(has_networks)

    def test_inventory_item_removed(self):
        inventory = self.inv

        # Make sure we have log_hosts in the original inventory
        self.assertIn('log_hosts', inventory)

        mi.remove_inventory_item("log_hosts", inventory)
        mi.remove_inventory_item("log_hosts", inventory, TARGET_DIR)

        # Now make sure it's gone
        self.assertIn('log_hosts', inventory)

    def test_metal_ips_kept(self):
        mi.remove_ip_addresses(self.inv)
        hostvars = self.inv['_meta']['hostvars']

        for host, variables in hostvars.items():
            has_networks = 'container_networks' in variables
            if not variables.get('is_metal', False):
                continue
            self.assertTrue(has_networks)

    def test_ansible_host_vars_removed(self):
        mi.remove_ip_addresses(self.inv)
        hostvars = self.inv['_meta']['hostvars']

        for host, variables in hostvars.items():
            has_host = 'ansible_host' in variables
            if variables.get('is_metal', False):
                continue
            self.assertFalse(has_host)

    def test_multiple_calls(self):
        """Removal should fail silently if keys are absent."""
        mi.remove_ip_addresses(self.inv)
        mi.remove_ip_addresses(self.inv)


if __name__ == '__main__':
    unittest.main(catchbreak=True)




===========Repository Name===========
openstack-ansible-ops
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\openstack-ansible-ops\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 2.0
skipsdist = True
envlist = docs,linters,functional


[testenv]
usedevelop = True
install_command =
    pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
commands =
    /usr/bin/find . -type f -name "*.pyc" -delete
passenv =
    COMMON_TESTS_PATH
    HOME
    http_proxy
    HTTP_PROXY
    https_proxy
    HTTPS_PROXY
    no_proxy
    NO_PROXY
    TESTING_BRANCH
    TESTING_HOME
    USER
whitelist_externals =
    bash
setenv =
    VIRTUAL_ENV={envdir}
    WORKING_DIR={toxinidir}


[testenv:docs]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands=
    bash -c "rm -rf doc/build"
    doc8 doc
    sphinx-build -b html doc/source doc/build/html


[doc8]
# Settings for doc8:
extensions = .rst


[testenv:releasenotes]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands =
    sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html


# environment used by the -infra templated docs job
[testenv:venv]
basepython = python3
commands =
    {posargs}


[testenv:pep8]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-pep8.sh"


[flake8]
# Ignores the following rules due to how ansible modules work in general
#     F403 'from ansible.module_utils.basic import *' used;
#          unable to detect undefined names
ignore=F403


[testenv:bashate]
commands =
    bash -c "{toxinidir}/tests/common/test-bashate.sh"


[testenv:linters]
basepython = python3
commands =
    bash -c "{toxinidir}/tests/common/test-ansible-env-prep.sh"
    {[testenv:pep8]commands}
    {[testenv:bashate]commands}




===========Repository Name===========
tripleo-quickstart
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\tripleo-quickstart\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 1.6
envlist = docs, linters
skipdist = True

[testenv]
usedevelop = True
install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
setenv = VIRTUAL_ENV={envdir}
deps = -r{toxinidir}/test-requirements.txt
whitelist_externals =
    bash
    echo

[testenv:bindep]
basepython = python3
# Do not install any requirements. We want this to be fast and work even if
# system dependencies are missing, since it's used to tell you what system
# dependencies are missing! This also means that bindep must be installed
# separately, outside of the requirements files.
deps = bindep
commands = bindep test

[testenv:docs]
basepython = python3
commands = python setup.py build_sphinx

[testenv:bashate]
commands =
# Run bashate check for all bash scripts
# Ignores the following rules:
# E006: Line longer than 79 columns (as many scripts use jinja
#       templating, this is very difficult)
# E040: Syntax error determined using `bash -n` (as many scripts
#       use jinja templating, this will often fail and the syntax
#       error will be discovered in execution anyway)
    bash -c "git ls-files | xargs grep --binary-files=without-match \
        --files-with-match '^.!.*\(ba\)\?sh$' \
        --exclude-dir .tox \
        --exclude-dir .git \
        | xargs bashate --error . --verbose --ignore=E006,E040"

[testenv:pep8]
basepython = python3
commands =
    # Run hacking/flake8 check for all python files
    bash -c "git ls-files | xargs grep --binary-files=without-match \
        --files-with-match '^.!.*python$' \
        --exclude-dir .tox \
        --exclude-dir .git \
        --exclude-dir .eggs \
        --exclude-dir *.egg-info \
        --exclude-dir dist \
        --exclude-dir *lib/python* \
        --exclude-dir doc \
        | xargs flake8 --verbose"

[testenv:ansible-lint]
basepython=python2
commands =
  bash ci-scripts/ansible-lint.sh

[testenv:linters]
basepython = python3
commands =
    python -m yamllint .
    {[testenv:bashate]commands}
    {[testenv:pep8]commands}
    {[testenv:ansible-lint]commands}

[testenv:releasenotes]
basepython = python3
whitelist_externals = bash
commands = bash -c ci-scripts/releasenotes_tox.sh

[testenv:venv]
changedir = {toxinidir}
commands =
    {posargs:echo done}

[flake8]
# E123, E125 skipped as they are invalid PEP-8.
show-source = True
ignore = E123,E125
builtins = _




===========Repository Name===========
windmill
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\windmill\tox.ini
===========File Type===========
.ini
===========File Content===========
[tox]
minversion = 1.4.2
envlist = docs,linters
skipsdist = True

[testenv]
deps = -r{toxinidir}/requirements.txt
       -r{toxinidir}/test-requirements.txt

[testenv:bindep]
basepython = python3
# Do not install any requirements. We want this to be fast and work even if
# system dependencies are missing, since it's used to tell you what system
# dependencies are missing! This also means that bindep must be installed
# separately, outside of the requirements files.
deps = bindep
commands =
  {toxinidir}/tools/install_bindep.sh

[testenv:docs]
basepython = python3
commands = python setup.py build_sphinx

[testenv:linters]
basepython = python3
whitelist_externals = bash
commands =
  flake8
  bash -c "cd playbooks; find . -type f -regex '.*.y[a]?ml' -print0 | xargs -t -n1 -0 \
    ansible-lint"

[testenv:venv]
basepython = python3
commands = {posargs}
passenv =
  HOME
  SSH_AUTH_SOCK
  TERM
  USER
setenv =
  ANSIBLE_CALLBACK_PLUGINS = {envsitepackagesdir}/ara/plugins/callbacks
  ANSIBLE_CONFIG = {toxinidir}/tests/ansible.cfg
  PYTHONUNBUFFERED = 1

[flake8]
# E123, E125 skipped as they are invalid PEP-8.

show-source = True
ignore = E123,E125
builtins = _
exclude=.venv,.git,.tox,dist,doc,*openstack/common*,*lib/python*,*egg,build




===========Repository Name===========
windmill
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\windmill\tests\ansible.cfg
===========File Type===========
.cfg
===========File Content===========
[defaults]
callback_whitelist = profile_tasks, timer

[ssh_connection]
# NOTE(pabelanger): Enable pipelining to deal with becomes issues:
# http://docs.ansible.com/ansible/become.html#becoming-an-unprivileged-user
pipelining = True




===========Repository Name===========
windmill
===========File Path===========
C:\Users\mehedi.md.hasan\PythonWorkspace\OSTK_ANSI\ostk-ansi\windmill\tests\collect-logs.yaml
===========File Type===========
.yaml
===========File Content===========
- hosts: all,!bastion
  tasks:
    - name: Ensure journald logs directory exists
      file:
        path: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal"
        state: directory

- hosts: statsd01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - statsd

    - name: Collect statsd log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/statsd
        - /var/log/statsd

- hosts: gear01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - gear

    - name: Collect gear log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/gear/logging.conf
        - /var/log/gear

- hosts: nb01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - nodepool-builder

    - name: Collect nodepool-builder log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/nodepool/builder-logging.conf
        - /etc/nodepool/nodepool.yaml
        - /var/log/nodepool/builds
        - /var/log/nodepool/builder-debug.log
        - /var/log/nodepool/nodepool-builder.log

- hosts: nl01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - nodepool-launcher

    - name: Collect nodepool-launcher log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/nodepool/launcher-logging.conf
        - /etc/nodepool/nodepool.yaml
        - /var/log/nodepool/launcher-debug.log
        - /var/log/nodepool/nodepool-launcher.log

- hosts: zs01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - zuul-scheduler

    - name: Collect zuul-scheduler log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/zuul/main.yaml
        - /etc/zuul/scheduler-logging.conf
        - /etc/zuul/zuul.conf
        - /var/log/zuul/scheduler-debug.log
        - /var/log/zuul/scheduler.log

- hosts: ze01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - zuul-executor

    - name: Collect zuul-executor log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/zuul/executor-logging.conf
        - /etc/zuul/zuul.conf
        - /var/log/zuul/executor-debug.log
        - /var/log/zuul/executor.log

- hosts: zf01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - zuul-fingergw

    - name: Collect zuul-fingergw log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/zuul/fingergw-logging.conf
        - /etc/zuul/zuul.conf
        - /var/log/zuul/fingergw-debug.log
        - /var/log/zuul/fingergw.log

- hosts: zm01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - zuul-merger

    - name: Collect zuul-merger log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/zuul/merger-logging.conf
        - /etc/zuul/zuul.conf
        - /var/log/zuul/merger-debug.log
        - /var/log/zuul/merger.log

- hosts: zw01
  tasks:
    - name: Collect journald logs
      shell: "sudo journalctl -u {{ item }}.service | tee {{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      args:
        creates: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}/var/log/journal/{{ item }}.service.log"
      with_items:
        - zuul-web

    - name: Collect zuul-web log files
      become: yes
      synchronize:
        dest: "{{ zuul_output_dir }}/logs/logs/{{ inventory_hostname }}"
        rsync_opts:
          - "--relative"
          - "--chown={{ ansible_user_id }}:{{ ansible_user_id }}"
        src: "{{ item }}"
        verify_host: true
      delegate_to: "{{ inventory_hostname }}"
      with_items:
        - /etc/zuul/web-logging.conf
        - /etc/zuul/zuul.conf
        - /var/log/zuul/web-debug.log
        - /var/log/zuul/web.log



------------> Mishandled privilege escalation: Use of sudo instead of using 'become' (https://docs.ansible.com/ansible/2.4/become.html)