<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-

# This work was created by participants in the DataONE project, and is
# jointly copyrighted by participating institutions in DataONE. For
# more information on DataONE, see our web site at http://dataone.org.
#
#   Copyright ${year}
#
# 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.
'''
:mod:`session`
==============

:Synopsis: Hold and manipulate session parameters.
:Created: 2011-11-20
:Author: DataONE (Dahl)
'''

# Stdlib.
import ast
import copy
import logging
import os
import pickle
import pprint
import ConfigParser
from types import *

# D1.
import d1_common.const
from d1_common.util import get_checksum_calculator_by_dataone_designator

# App.
from print_level import *
import cli_exceptions
import system_metadata
import access_control
import replication_policy

# Session variable mapping.
#
session_variable_dict = {
  'dataoneurl': ('dataone-url', str),
  'mnurl': ('mn-url', str),
  'certpath': ('cert-file', str),
  'keypath': ('key-file', str),
  'objectformat': ('object-format', str),
  'rightsholder': ('rights-holder', str),
  'originmn': ('origin-mn', str),
  'authoritativemn': ('authoritative-mn', str),
  'fromdate': ('from-date', str),
  'todate': ('to-date', str),
  'searchobjectformat': ('search-object-format', str),
  'querytype': ('query-type', str),
}


class session(object):
  def __init__(self):
    self.session = self.get_default_session()
    self.access_control = access_control.access_control()
    self.replication_policy = replication_policy.replication_policy()

  def reset(self):
    self.__init__()

  def get_default_session(self):
    return copy.deepcopy({
      'cli': {
        'pretty': (True, bool),
        'verbose': (False, bool),
      },
      'node': {
        'dataone-url': (d1_common.const.URL_DATAONE_ROOT, str),
        'mn-url': ('https://localhost/mn/', str),
      },
      'slice': {
        'start': (0, int),
        'count': (d1_common.const.MAX_LISTOBJECTS, int),
      },
      'auth': {
        'anonymous': (True, bool),
        'cert-file': (None, str),
        'key-file': (None, str),
      },
      'sysmeta': {
        'object-format': (None, str),
        'submitter': (None, str),
        'rights-holder': (None, str),
        'origin-mn': (None, str),
        'authoritative-mn': (None, str),
        'algorithm': (d1_common.const.DEFAULT_CHECKSUM_ALGORITHM, str),
      },
      'search': {
        'from-date': (None, str),
        'to-date': (None, str),
        'search-object-format': (None, str),
        'query-type': ('solr', str),
        'query': ('*:*', str),
      },
    })

  def get_session_section_ordering(self):
    return 'cli', 'node', 'slice', 'auth', 'sysmeta', 'search'

  def get_default_pickle_file_path(self):
    return os.path.join(os.environ['HOME'], '.d1client.conf')

  def _find_section_containing_session_parameter(self, name):
    '''Find the section containing a session parameter.'''
    # Because the session parameters are split into sections and the user does
    # not specify the section when setting a parameter, it is necessary to
    # search for the parameter.
    for section in self.session:
      if name in self.session[section]:
        return section
    raise cli_exceptions.InvalidArguments('Invalid session parameter: {0}'.format(name))

  def _assert_valid_session_parameter(self, section, name):
    try:
      self.session[section][name]
    except LookupError:
      raise cli_exceptions.InvalidArguments(
        'Invalid session parameter: {0} / {1}'.format(section, name)
      )

  #=============================================================================
  # Public.
  #=============================================================================

  def get(self, section, name):
    self._assert_valid_session_parameter(section, name)
    return self.session[section][name][0]

  def get_with_implicit_section(self, name):
    section = self._find_section_containing_session_parameter(name)
    return self.get(section, name)

  def set(self, section, name, value):
    self._assert_valid_session_parameter(section, name)
    name_type = self.session[section][name][1]
    self.session[section][name] = (value, name_type)

  def set_with_implicit_section(self, name, value):
    section = self._find_section_containing_session_parameter(name)
    self.set(section, name, value)

  def set_with_conversion(self, section, name, value_string):
    '''Convert user supplied string to Python type. Lets user use values such as
    True, False and integers. All parameters can be set to None, regardless of
    type. Handle the case where a string is typed by the user and is not quoted,
    as a string literal.
    '''
    self._assert_valid_session_parameter(section, name)
    try:
      v = ast.literal_eval(value_string)
    except (ValueError, SyntaxError):
      v = value_string
    if v is None:
      self.set(section, name, None)
    else:
      try:
        type_converter = self.session[section][name][1]
        value_string = self.validate_value_type(value_string, type_converter)
        self.set(section, name, type_converter(v))
      except ValueError as e:
        raise cli_exceptions.InvalidArguments(
          'Invalid value for {0} / {1}: {2}'.format(section, name, value_string)
        )

  def validate_value_type(self, value_string, type_converter):
    # Make sure booleans are "sane"
    if type_converter is BooleanType:
      if value_string in ('true', 'True', 't', 'T', 1, 'yes', 'Yes'):
        return 'True'
      elif value_string in ('false', 'False', 'f', 'F', 0, 'no', 'No'):
        return 'False'
      else:
        raise ValueError('"%s": Invalid boolean value' % value_string)

  def set_with_conversion_implicit_section(self, name, value_string):
    section = self._find_section_containing_session_parameter(name)
    self.set_with_conversion(section, name, value_string)

  #=============================================================================
  # Access control.
  #=============================================================================

  def access_control_add_allowed_subject(self, subject, permission):
    self.access_control.add_allowed_subject(subject, permission)

  def access_control_remove_allowed_subject(self, subject):
    self.access_control.remove_allowed_subject(subject)

  def access_control_allow_public(self, allow):
    self.access_control.allow_public(allow)

  def access_control_remove_all_allowed_subjects(self):
    self.access_control.remove_all_allowed_subjects()

  def access_control_get_pyxb(self):
    return self.access_control.to_pyxb()
<commit_msg>test_110 - Allow '0'/'1' (strings) as well as 0/1 (numbers).<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-

# This work was created by participants in the DataONE project, and is
# jointly copyrighted by participating institutions in DataONE. For
# more information on DataONE, see our web site at http://dataone.org.
#
#   Copyright ${year}
#
# 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.
'''
:mod:`session`
==============

:Synopsis: Hold and manipulate session parameters.
:Created: 2011-11-20
:Author: DataONE (Dahl)
'''

# Stdlib.
import ast
import copy
import logging
import os
import pickle
import pprint
import ConfigParser
from types import *

# D1.
import d1_common.const
from d1_common.util import get_checksum_calculator_by_dataone_designator

# App.
from print_level import *
import cli_exceptions
import system_metadata
import access_control
import replication_policy

# Session variable mapping.
#
session_variable_dict = {
  'dataoneurl': ('dataone-url', str),
  'mnurl': ('mn-url', str),
  'certpath': ('cert-file', str),
  'keypath': ('key-file', str),
  'objectformat': ('object-format', str),
  'rightsholder': ('rights-holder', str),
  'originmn': ('origin-mn', str),
  'authoritativemn': ('authoritative-mn', str),
  'fromdate': ('from-date', str),
  'todate': ('to-date', str),
  'searchobjectformat': ('search-object-format', str),
  'querytype': ('query-type', str),
}


class session(object):
  def __init__(self):
    self.session = self.get_default_session()
    self.access_control = access_control.access_control()
    self.replication_policy = replication_policy.replication_policy()

  def reset(self):
    self.__init__()

  def get_default_session(self):
    return copy.deepcopy({
      'cli': {
        'pretty': (True, bool),
        'verbose': (False, bool),
      },
      'node': {
        'dataone-url': (d1_common.const.URL_DATAONE_ROOT, str),
        'mn-url': ('https://localhost/mn/', str),
      },
      'slice': {
        'start': (0, int),
        'count': (d1_common.const.MAX_LISTOBJECTS, int),
      },
      'auth': {
        'anonymous': (True, bool),
        'cert-file': (None, str),
        'key-file': (None, str),
      },
      'sysmeta': {
        'object-format': (None, str),
        'submitter': (None, str),
        'rights-holder': (None, str),
        'origin-mn': (None, str),
        'authoritative-mn': (None, str),
        'algorithm': (d1_common.const.DEFAULT_CHECKSUM_ALGORITHM, str),
      },
      'search': {
        'from-date': (None, str),
        'to-date': (None, str),
        'search-object-format': (None, str),
        'query-type': ('solr', str),
        'query': ('*:*', str),
      },
    })

  def get_session_section_ordering(self):
    return 'cli', 'node', 'slice', 'auth', 'sysmeta', 'search'

  def get_default_pickle_file_path(self):
    return os.path.join(os.environ['HOME'], '.d1client.conf')

  def _find_section_containing_session_parameter(self, name):
    '''Find the section containing a session parameter.'''
    # Because the session parameters are split into sections and the user does
    # not specify the section when setting a parameter, it is necessary to
    # search for the parameter.
    for section in self.session:
      if name in self.session[section]:
        return section
    raise cli_exceptions.InvalidArguments('Invalid session parameter: {0}'.format(name))

  def _assert_valid_session_parameter(self, section, name):
    try:
      self.session[section][name]
    except LookupError:
      raise cli_exceptions.InvalidArguments(
        'Invalid session parameter: {0} / {1}'.format(section, name)
      )

  #=============================================================================
  # Public.
  #=============================================================================

  def get(self, section, name):
    self._assert_valid_session_parameter(section, name)
    return self.session[section][name][0]

  def get_with_implicit_section(self, name):
    section = self._find_section_containing_session_parameter(name)
    return self.get(section, name)

  def set(self, section, name, value):
    self._assert_valid_session_parameter(section, name)
    name_type = self.session[section][name][1]
    self.session[section][name] = (value, name_type)

  def set_with_implicit_section(self, name, value):
    section = self._find_section_containing_session_parameter(name)
    self.set(section, name, value)

  def set_with_conversion(self, section, name, value_string):
    '''Convert user supplied string to Python type. Lets user use values such as
    True, False and integers. All parameters can be set to None, regardless of
    type. Handle the case where a string is typed by the user and is not quoted,
    as a string literal.
    '''
    self._assert_valid_session_parameter(section, name)
    try:
      v = ast.literal_eval(value_string)
    except (ValueError, SyntaxError):
      v = value_string
    if v is None:
      self.set(section, name, None)
    else:
      try:
        type_converter = self.session[section][name][1]
        value_string = self.validate_value_type(value_string, type_converter)
        self.set(section, name, type_converter(v))
      except ValueError as e:
        raise cli_exceptions.InvalidArguments(
          'Invalid value for {0} / {1}: {2}'.format(section, name, value_string)
        )

  def validate_value_type(self, value_string, type_converter):
    # Make sure booleans are "sane"
    if type_converter is BooleanType:
      if value_string in ('true', 'True', 't', 'T', 1, '1', 'yes', 'Yes'):
        return 'True'
      elif value_string in ('false', 'False', 'f', 'F', 0, '0', 'no', 'No'):
        return 'False'
      else:
        raise ValueError('"%s": Invalid boolean value' % value_string)

  def set_with_conversion_implicit_section(self, name, value_string):
    section = self._find_section_containing_session_parameter(name)
    self.set_with_conversion(section, name, value_string)

  #=============================================================================
  # Access control.
  #=============================================================================

  def access_control_add_allowed_subject(self, subject, permission):
    self.access_control.add_allowed_subject(subject, permission)

  def access_control_remove_allowed_subject(self, subject):
    self.access_control.remove_allowed_subject(subject)

  def access_control_allow_public(self, allow):
    self.access_control.allow_public(allow)

  def access_control_remove_all_allowed_subjects(self):
    self.access_control.remove_all_allowed_subjects()

  def access_control_get_pyxb(self):
    return self.access_control.to_pyxb()

{'fraction_non_alphanumeric': 0.16350100883601196, 'fraction_numerical': 0.005078967508522925, 'mean_word_length': 3.915868673050616, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2128196', 'n_tokens_mistral': 4357, 'n_tokens_neox': 4226, 'n_words': 1158}