text:
from ast import literal_eval
from datetime import datetime
from os import environ
import pytz

from babel import Locale, UnknownLocaleError
from pycountry import languages

from beaker.cache import cache_region

from markdown import markdown
import colander
from deform import ValidationFailure

from pyramid.view import view_config
from pyramid.renderers import get_renderer
from pyramid.decorator import reify
from pyramid.response import Response
from pyramid.security import forget, remember
from pyramid.httpexceptions import HTTPFound, HTTPNotFound, HTTPBadRequest

from elasticgit import F

from cms import USER_DATA_SESSION_KEY
from cms.views.base import BaseCmsView
from cms.views.utils import (
    EGPaginator, to_eg_objects, translation_string_factory as _, ga_context)
from cms.views.forms import CommentForm
from cms.tasks import pull

from unicore.content.models import Category, Page, Localisation
from unicore.hub.client import ClientException as HubClientException
from unicore.hub.client.utils import same_origin
from unicore.comments.client import (
    LazyCommentPage, UserBanned, CommentStreamNotOpen, CommentServiceException)

from pyramid.view import notfound_view_config

CACHE_TIME = 'default_term'
COMMENTS_PER_PAGE = 20

class CmsViews(BaseCmsView):

    @reify
    def get_available_languages(self):
        available_languages = sorted(literal_eval(
            (self.settings.get('available_languages', '[]'))),
            key=lambda tup: tup[1].lower())
        return [
            (code, self.get_display_name(code))
            for code, name in available_languages]

    @reify
    def get_featured_languages(self):
        featured_languages = sorted(literal_eval(
            (self.settings.get('featured_languages', '[]'))),
            key=lambda tup: tup[1].lower())
        return [
            (code, self.get_display_name(code))
            for code, name in featured_languages]

    def get_display_name(self, locale):
        language_code, _, country_code = locale.partition('_')
        term_code = languages.get(bibliographic=language_code).terminology

        available_languages = dict(literal_eval(
            (self.settings.get('available_languages', '[]'))))

        try:
            return Locale.parse(term_code).language_name
        except UnknownLocaleError:
            # Fallback value is the generated value in English or the code
            return available_languages.get(locale, locale)

    def get_display_languages(self):
        to_display = [
            code for code, name in
            self.get_featured_languages or self.get_available_languages[:2]]

        featured_and_current = [self.locale] + sorted(list(
            set(to_display) - set([self.locale])),
            key=lambda tup: tup[1].lower())
        return [
            (code, self.get_display_name(code))
            for code in featured_and_current]

    @reify
    def global_template(self):
        renderer = get_renderer("cms:templates/base.pt")
        return renderer.implementation().macros['layout']

    @reify
    def paginator_template_category(self):
        renderer = get_renderer("cms:templates/paginator_category.pt")
        return renderer.implementation().macros['paginator']

    @reify
    def paginator_template(self):
        renderer = get_renderer("cms:templates/paginator.pt")
        return renderer.implementation().macros['paginator']

    @reify
    def search_box_template(self):
        renderer = get_renderer("cms:templates/search_box.pt")
        return renderer.implementation().macros['search_box']

    @reify
    def logo_template(self):
        renderer = get_renderer("cms:templates/logo.pt")
        return renderer.implementation().macros['logo']

    @reify
    def auth_template(self):
        renderer = get_renderer("cms:templates/auth.pt")
        return renderer.implementation().macros['auth']

    def get_logo_attributes(self, default_image_src=None,
                            width=None, height=None):
        attrs = {'width': width, 'height': height}
        localisation = self.get_localisation()

        if not localisation:
            attrs.update({'src': default_image_src, 'alt': None})
            return attrs

        attrs['alt'] = localisation.logo_description or None
        if localisation.logo_image:
            attrs['src'] = self.get_image_url(localisation.logo_image_host,
                                              localisation.logo_image)
        else:
            attrs['src'] = default_image_src
        return attrs

    def get_localisation(self):
        try:
            [localisation] = self.workspace.S(
                Localisation).filter(locale=self.locale)
            return localisation.to_object()
        except ValueError:
            return None

    def get_categories(self, order_by=('position',)):
        return self._get_categories(self.locale, order_by)

    @cache_region(CACHE_TIME)
    def _get_categories(self, locale, order_by):
        return to_eg_objects(self.workspace.S(Category).filter(
            language=locale).order_by(*order_by))

    @cache_region(CACHE_TIME)
    def get_category(self, uuid):
        try:
            [category] = self.workspace.S(Category).filter(uuid=uuid)
            return category.to_object()
        except ValueError:
            return None

    def get_pages(self, limit=5, order_by=('position', '-modified_at')):
        """
        Return pages the GitModel knows about.
        :param int limit:
            The number of pages to return, defaults to 5.
        :param tuple order_by:
            The attributes to order on,
            defaults to ('position', '-modified_at')
        """
        return to_eg_objects(self.workspace.S(Page).filter(
            language=self.locale).order_by(*order_by)[:limit])

    @cache_region(CACHE_TIME)
    def _get_featured_pages(self, locale, limit, order_by):
        return to_eg_objects(self.workspace.S(Page).filter(
            language=locale, featured=True).order_by(*order_by)[:limit])

    def get_featured_pages(
            self, limit=5, order_by=('position', '-modified_at')):
        """
        Return featured pages the GitModel knows about.
        :param str locale:
            The locale string, like `eng_UK`.
        :param int limit:
            The number of pages to return, defaults to 5.
        :param tuple order_by:
            The attributes to order on,
            defaults to ('position', '-modified_at').
        """
        return self._get_featured_pages(self.locale, limit, order_by)

    @cache_region(CACHE_TIME)
    def get_pages_for_category(
            self, category_id, locale, order_by=('position',)):
        return self.workspace.S(Page)[:10000].filter(
            primary_category=category_id,
            language=locale).order_by(*order_by)

    def get_featured_category_pages(
            self, category_id, order_by=('position',)):
        return self._get_featured_category_pages(
            category_id, self.locale, order_by)

    @cache_region(CACHE_TIME)
    def _get_featured_category_pages(self, category_id, locale, order_by):
        return to_eg_objects(self.workspace.S(Page).filter(
            primary_category=category_id, language=locale,
            featured_in_category=True).order_by(*order_by))

    @cache_region(CACHE_TIME)
    def get_page(self, uuid=None, slug=None, locale=None):
        try:
            query = self.workspace.S(Page).filter(
                F(uuid=uuid) | F(slug=slug))
            if locale is not None:
                query = query.filter(language=locale)
            [page] = query[:1]
            return page.to_object()
        except ValueError:
            return None

    def get_comments_for_content(self, content_uuid, **page_args):
        commentclient = self.request.registry.commentclient
        if commentclient is None:
            return None

        default_page_args = {'limit': COMMENTS_PER_PAGE}

        if 'c_after' in self.request.GET:
            default_page_args['after'] = self.request.GET['c_after']
        elif 'c_before' in self.request.GET:
            default_page_args['before'] = self.request.GET['c_before']

        default_page_args.update(page_args)

        return LazyCommentPage(
            commentclient,
            content_uuid=content_uuid,
            app_uuid=commentclient.settings['app_id'],
            **default_page_args)

    def get_comment_context(self, content_object):
        return {
            'comments': self.get_comments_for_content(
                content_object.uuid, limit=COMMENTS_PER_PAGE),
            'comment_form': CommentForm(self.request, content_object)
        }

    @reify
    def get_top_nav(self, order_by=('position',)):
        return self._get_top_nav(self.locale, order_by)

    @cache_region(CACHE_TIME)
    def _get_top_nav(self, locale, order_by):
        return to_eg_objects(self.workspace.S(Category).filter(
            language=locale,
            featured_in_navbar=True).order_by(*order_by))

    @view_config(route_name='health', renderer='json')
    def health(self):
        app_id = environ.get('MARATHON_APP_ID', None)
        ver = environ.get('MARATHON_APP_VERSION', None)
        return {'id': app_id, 'version': ver}

    @ga_context(lambda context: {'dt': 'Home', })
    @view_config(route_name='home', renderer='cms:templates/home.pt')
    @view_config(route_name='home_jinja', renderer='cms:templates/home.jinja2')
    # redundantcategory
    @view_config(route_name='categories',
                 renderer='cms:templates/categories.pt')
    def categories(self):
        return {}

    @ga_context(lambda context: {'dt': context['category'].title, })
    @view_config(route_name='category', renderer='cms:templates/category.pt')
    @view_config(route_name='category_jinja2',
                 renderer='cms:templates/category.jinja2')
    def category(self):
        category_id = self.request.matchdict['category']
        category = self.get_category(category_id)

        if category is None or category.language != self.locale:
            raise HTTPNotFound()

        pages = self.get_pages_for_category(category_id, self.locale)
        p = int(self.request.GET.get('p', 0))
        empty_defaults = {
            'paginator': [],
            'category': category,
            'p': p,
        }
        if not category:
            return empty_defaults

        if len(pages) == 0:
            return empty_defaults
        paginator = EGPaginator(
            pages, p, results_per_page=self.results_per_page)

        # requested page number is out of range
        total_pages = paginator.total_pages()
        # sets the floor to 0
        p = p if p >= 0 else 0
        # sets the roof to `total_pages -1`
        p = p if p < total_pages else total_pages - 1
        paginator = EGPaginator(
            pages, p, results_per_page=self.results_per_page)

        return {
            'paginator': paginator,
            'category': category,
            'p': p,
            'pages': pages,
        }

    @ga_context(lambda context: {'dt': context['page'].title, })
    @view_config(route_name='content', renderer='cms:templates/content.pt')
    @view_config(route_name='content_jinja',
                 renderer='cms:templates/content.jinja2')
    def content(self):
        page = self.get_page(self.request.matchdict['uuid'])

        if not page:
            raise HTTPNotFound()

        if page.linked_pages:
            linked_pages = to_eg_objects(self.workspace.S(Page).filter(
                uuid__in=page.linked_pages))
        else:
            linked_pages = []

        category = None
        if page.primary_category:
            category = self.get_category(page.primary_category)

        if page.language != self.locale:
            raise HTTPNotFound()

        context = {
            'page': page,
            'linked_pages': linked_pages,
            'primary_category': category,
            'content': markdown(page.content),
            'description': markdown(page.description),
        }
        context.update(self.get_comment_context(page))
        return context

    @view_config(route_name='comments',
                 renderer='cms:templates/comments/comment_page.jinja2')
    def comments(self):
        context = self.content()

        if 'submit' in self.request.POST:
            form = context['comment_form']

            try:
                data = form.validate(self.request.POST.items())
                commentclient = self.request.registry.commentclient
                commentclient.create_comment(data)
                raise HTTPFound(self.request.route_url(
                    'comments', uuid=data['content_uuid']))

            except ValidationFailure as e:
                context['comment_form'] = e.field

            except UserBanned:
                form.error = colander.Invalid(
                    form, _('You have been banned from commenting'))

            except CommentStreamNotOpen:
                raise HTTPFound(self.request.route_url(
                    'comments', uuid=data['content_uuid']))

            except ValueError as e:
                raise HTTPBadRequest()

        return context

    @view_config(route_name='flag_comment')
    def flag_comment(self):
        commentclient = self.request.registry.commentclient

        if None in (self.request.user, commentclient):
            raise HTTPNotFound

        flag_data = {
            'user_uuid': self.request.user.get('uuid'),
            'comment_uuid': self.request.matchdict['uuid'],
            'submit_datetime': datetime.now(pytz.utc).isoformat(),
            'app_uuid': commentclient.settings['app_id']
        }
        try:
            commentclient.create_flag(flag_data)
        except CommentServiceException as e:
            if e.response.status_code == 404:
                raise HTTPNotFound
            raise e

        query = {}
        if self.request.referrer and same_origin(
                self.request.referrer, self.request.current_route_url()):
            query = {'next': self.request.referrer}

        return HTTPFound(self.request.route_url(
            'flag_comment_success', uuid=flag_data['comment_uuid'],
            _query=query))

    @view_config(route_name='flag_comment_success',
                 renderer='cms:templates/comments/comment_flagged.jinja2')
    def flag_comment_success(self):
        if not self.request.user:
            raise HTTPNotFound

        next_url = self.request.GET.get('next')
        if next_url and not same_origin(
                next_url, self.request.current_route_url()):
            next_url = None

        return {'next': next_url}

    @ga_context(lambda context: {'dt': context['page'].title, })
    @view_config(route_name='flatpage', renderer='cms:templates/flatpage.pt')
    @view_config(route_name='flatpage_jinja',
                 renderer='cms:templates/flatpage.jinja2')
    def flatpage(self):
        page = self.get_page(
            None, self.request.matchdict['slug'], self.locale)

        if not page:
            raise HTTPNotFound()

        if page.language != self.locale:
            raise HTTPNotFound()

        return {
            'page': page,
            'content': markdown(page.content),
            'description': markdown(page.description),
        }

    @ga_context(lambda context: {'dt': 'Choose Language', })
    @view_config(
        route_name='locale_change',
        renderer='cms:templates/locale_change.pt')
    @view_config(
        route_name='locale_change_jinja',
        renderer='cms:templates/locale_change.jinja2')
    def locale_change(self):
        return {
            'languages': self.get_featured_languages +
            sorted(list(set(self.get_available_languages) -
                        set(self.get_featured_languages)),
                   key=lambda tup: tup[1].lower())
        }

    @ga_context(lambda context: {'dt': 'Set Language', })
    @view_config(route_name='locale')
    @view_config(route_name='locale_matched')
    def set_locale_cookie(self):
        response = Response()
        language = self.request.matchdict.get('language') or \
            self.request.GET.get('language')

        if language:
            response.set_cookie('_LOCALE_', value=language, max_age=31536000)

        return HTTPFound(location='/', headers=response.headers)

    @ga_context(lambda context: {'dt': 'Search', })
    @view_config(route_name='search',
                 renderer='cms:templates/search_results.pt')
    @view_config(route_name='search_jinja',
                 renderer='cms:templates/search_results.jinja2')
    def search(self):

        query = self.request.GET.get('q')
        p = int(self.request.GET.get('p', 0))

        empty_defaults = {
            'paginator': [],
            'query': query,
            'p': p,
        }

        # handle query exception
        if not query:
            return empty_defaults

        all_results = self.workspace.S(Page).query(
            content__query_string=query).filter(language=self.locale)

        # no results found
        if all_results.count() == 0:
            return empty_defaults

        paginator = EGPaginator(all_results, p)

        # requested page number is out of range
        total_pages = paginator.total_pages()
        # sets the floor to 0
        p = p if p >= 0 else 0
        # sets the roof to `total_pages -1`
        p = p if p < total_pages else total_pages - 1
        paginator = EGPaginator(all_results, p)

        return {
            'paginator': paginator,
            'query': query,
            'p': p,
        }

    @notfound_view_config(renderer='cms:templates/404.pt')
    def notfound(self):
        self.request.response.status = 404
        return {}

    @view_config(route_name='login')
    def login(self):
        hubclient = self.request.registry.hubclient
        response = HTTPFound()

        # redeem ticket to get user data
        ticket = self.request.GET.get('ticket', None)
        if ticket and hubclient:
            try:
                user = hubclient.get_user(
                    ticket, self.request.route_url('redirect_to_login'))
                self.request.session[USER_DATA_SESSION_KEY] = user.data
                user_id = user.get('uuid')
                headers = remember(self.request, user_id)
                response.headerlist.extend(headers)

            except HubClientException:
                # TODO: what to do when ticket is invalid?
                pass

        redirect_url = self.request.GET.get('url', None)
        if not (redirect_url and same_origin(
                redirect_url, self.request.current_route_url())):
            redirect_url = self.request.route_url(route_name='home')
        response.location = redirect_url

        return response

    @view_config(route_name='redirect_to_login')
    def redirect_to_login(self):
        hubclient = self.request.registry.hubclient

        if self.request.referrer and same_origin(
                self.request.referrer, self.request.current_route_url()):
            callback_url = self.request.route_url(
                route_name='login', _query={'url': self.request.referrer})
        else:
            callback_url = self.request.route_url(route_name='login')

        if hubclient is None:
            # benign redirect if hubclient is not configured
            return HTTPFound(callback_url)

        return HTTPFound(hubclient.get_login_redirect_url(
            callback_url, locale=self.locale))

    @view_config(route_name='logout')
    def logout(self):
        response = HTTPFound(headers=forget(self.request))

        if self.request.referrer and same_origin(
                self.request.referrer, self.request.current_route_url()):
            response.location = self.request.referrer
        else:
            response.location = self.request.route_url(route_name='home')

        return response

    @view_config(route_name='api_notify', renderer='json')
    def api_notify(self):
        pull.delay(
            self.settings['git.path'],
            index_prefix=self.settings['es.index_prefix'],
            es=self.es_settings)
        return {}

    @view_config(route_name='repos', renderer='json')
    def repos(self):
        index = self.settings['es.index_prefix']

        return [{
            'index': index,
            'data': {'name': index},
        }]


text:
<?php
include_once  "../models/Comment.php";

class CommentDao extends BaseDao{
     private $tableName = "comments";
     public function __construct() { }

     public function add($comment){
        $conn = get_connection();
        $query = 'INSERT INTO '.$this->tableName
           . ' (authorId, articleId, text, publishDate) '
           . 'VALUES ("' . $comment->getAuthorId() . '", "' . $comment->getArticleId() . '", "' . $comment->getText() . '", "' . date('Y-m-d', strtotime(date('Y-m-d'))) . '")';

        //$entity = mysqli_query($conn, $query);
        if (!mysqli_query($conn, $query)) {
              printf("Errormessage: %s\n", mysqli_error($conn));
        }

        mysqli_close($conn);
        //return $entity;
     }

     public function update($comment){
        $conn = get_connection();
        $query = 'UPDATE articles
                        SET
                        authorId = "' . $comment -> getAuthorId()
                        . '", articleId = "' . $comment -> getArticleId()
                        . '", text = "' . $comment -> getText()
                        . '", publishDate = "' . $comment -> getPublishDate()
                        . '" WHERE id = ' . $article->getId();

        $entity = mysqli_query($conn, $query);
        mysqli_close($conn);
        return $entity;
     }

     public function findById($id){
        // $conn = get_connection();
        // $query = 'SELECT * FROM '.$this->tableName.' WHERE id = ' . $id;

        // $entity = mysqli_query($conn, $query);
        // mysqli_close($conn);
        // return $entity;
        return null;
     }

     public function findAllByArticleId($articleId){
        $conn = get_connection();
        $userDao = new UserDao();
        $query = 'SELECT id, authorId, articleId, text, publishDate FROM '.$this->tableName . ' WHERE articleId = ' . $articleId . ';';

        $entitiesList = array();
        $dbQuery = mysqli_query($conn, $query);
        while( $row = mysqli_fetch_assoc($dbQuery) ){
            $entity = new Comment($row["id"], $row["authorId"], $row["articleId"], $row["text"], $row["publishDate"]);
            $author = $userDao->findById($entity->getAuthorId());
            $entity -> setAuthor($author);
            $entitiesList[] = $entity;
        }
        mysqli_close($conn);
        return $entitiesList;
     }

     public function findAllByFilter($skip, $take){
         return null;
     }

     public function deleteById($id){
        $conn = get_connection();
        $query = 'DELETE FROM '.$this->tableName.' WHERE id = ' . $id;
        mysqli_query($conn, $query);
        mysqli_close($conn);
     }

     public function deleteAll(){
        $conn = get_connection();
        $query = 'DELETE FROM '.$this->tableName;
        mysqli_query($conn, $query);
        mysqli_close($conn);
     }

}
?>

redpajama_set_name: RedPajamaGithub


text:
namespace gin {
class ContextHolder;
class IsolateHolder;
}

namespace extensions {

// A common unit test class for testing API bindings. Creates an isolate and an
// initial v8 context, and checks for v8 leaks at the end of the test.
class APIBindingTest : public testing::Test {
 protected:
  APIBindingTest();
  ~APIBindingTest() override;

  // testing::Test:
  void SetUp() override;
  void TearDown() override;

  v8::Local<v8::Context> ContextLocal();
  void DisposeContext();

  // Returns the associated isolate. Defined out-of-line to avoid the include
  // for IsolateHolder in the header.
  v8::Isolate* isolate();

 private:
  base::MessageLoop message_loop_;
  std::unique_ptr<gin::IsolateHolder> isolate_holder_;
  std::unique_ptr<gin::ContextHolder> context_holder_;

  DISALLOW_COPY_AND_ASSIGN(APIBindingTest);
};

}  // namespace extensions

#endif  // EXTENSIONS_RENDERER_API_BINDING_TEST_H_


text:
package org.jetbrains.jps.model.serialization.library;

import com.intellij.openapi.util.JDOMUtil;
import com.intellij.util.containers.MultiMap;
import org.jdom.Element;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.jps.model.*;
import org.jetbrains.jps.model.java.JpsJavaLibraryType;
import org.jetbrains.jps.model.library.JpsLibrary;
import org.jetbrains.jps.model.library.JpsLibraryCollection;
import org.jetbrains.jps.model.library.JpsLibraryRoot;
import org.jetbrains.jps.model.library.JpsOrderRootType;
import org.jetbrains.jps.model.module.JpsModuleReference;
import org.jetbrains.jps.model.serialization.JpsModelSerializerExtension;
import org.jetbrains.jps.model.serialization.JpsPathMapper;

public final class JpsLibraryTableSerializer {
  private static final JpsLibraryRootTypeSerializer[] PREDEFINED_ROOT_TYPES_SERIALIZERS = {
    new JpsLibraryRootTypeSerializer("CLASSES", JpsOrderRootType.COMPILED, true),
    new JpsLibraryRootTypeSerializer("SOURCES", JpsOrderRootType.SOURCES, true),
    new JpsLibraryRootTypeSerializer("DOCUMENTATION", JpsOrderRootType.DOCUMENTATION, false)
  };
  public static final String NAME_ATTRIBUTE = "name";
  public static final String TYPE_ATTRIBUTE = "type";
  public static final String PROPERTIES_TAG = "properties";
  public static final String JAR_DIRECTORY_TAG = "jarDirectory";
  private static final String URL_ATTRIBUTE = "url";
  public static final String ROOT_TAG = "root";
  public static final String RECURSIVE_ATTRIBUTE = "recursive";
  public static final String LIBRARY_TAG = "library";
  private static final JpsLibraryPropertiesSerializer<JpsDummyElement> JAVA_LIBRARY_PROPERTIES_SERIALIZER =
    new JpsLibraryPropertiesSerializer<JpsDummyElement>(JpsJavaLibraryType.INSTANCE, null) {
      @Override
      public JpsDummyElement loadProperties(@Nullable Element propertiesElement) {
        return JpsElementFactory.getInstance().createDummyElement();
      }
    };
  public static final String MODULE_LEVEL = "module";
  public static final String PROJECT_LEVEL = "project";
  public static final String APPLICATION_LEVEL = "application";

  public static void loadLibraries(@Nullable Element libraryTableElement,
                                   @NotNull JpsPathMapper pathMapper,
                                   JpsLibraryCollection result) {
    for (Element libraryElement : JDOMUtil.getChildren(libraryTableElement, LIBRARY_TAG)) {
      result.addLibrary(loadLibrary(libraryElement, pathMapper));
    }
  }

  public static JpsLibrary loadLibrary(Element libraryElement, @NotNull JpsPathMapper pathMapper) {
    return loadLibrary(libraryElement, libraryElement.getAttributeValue(NAME_ATTRIBUTE), pathMapper);
  }

  public static JpsLibrary loadLibrary(Element libraryElement, String name, @NotNull JpsPathMapper pathMapper) {
    String typeId = libraryElement.getAttributeValue(TYPE_ATTRIBUTE);
    final JpsLibraryPropertiesSerializer<?> loader = getLibraryPropertiesSerializer(typeId);
    JpsLibrary library = createLibrary(name, loader, libraryElement.getChild(PROPERTIES_TAG));

    MultiMap<JpsOrderRootType, String> jarDirectories = new MultiMap<>();
    MultiMap<JpsOrderRootType, String> recursiveJarDirectories = new MultiMap<>();
    for (Element jarDirectory : JDOMUtil.getChildren(libraryElement, JAR_DIRECTORY_TAG)) {
      String url = pathMapper.mapUrl(jarDirectory.getAttributeValue(URL_ATTRIBUTE));
      String rootTypeId = jarDirectory.getAttributeValue(TYPE_ATTRIBUTE);
      final JpsOrderRootType rootType = rootTypeId != null ? getRootType(rootTypeId) : JpsOrderRootType.COMPILED;
      boolean recursive = Boolean.parseBoolean(jarDirectory.getAttributeValue(RECURSIVE_ATTRIBUTE));
      jarDirectories.putValue(rootType, url);
      if (recursive) {
        recursiveJarDirectories.putValue(rootType, url);
      }
    }
    for (Element rootsElement : libraryElement.getChildren()) {
      final String rootTypeId = rootsElement.getName();
      if (!rootTypeId.equals(JAR_DIRECTORY_TAG) && !rootTypeId.equals(PROPERTIES_TAG) && !rootTypeId.equals("excluded")) {
        final JpsOrderRootType rootType = getRootType(rootTypeId);
        for (Element rootElement : JDOMUtil.getChildren(rootsElement, ROOT_TAG)) {
          String url = pathMapper.mapUrl(rootElement.getAttributeValue(URL_ATTRIBUTE));
          JpsLibraryRoot.InclusionOptions options;
          if (jarDirectories.get(rootType).contains(url)) {
            final boolean recursive = recursiveJarDirectories.get(rootType).contains(url);
            options = recursive ? JpsLibraryRoot.InclusionOptions.ARCHIVES_UNDER_ROOT_RECURSIVELY : JpsLibraryRoot.InclusionOptions.ARCHIVES_UNDER_ROOT;
          }
          else {
            options = JpsLibraryRoot.InclusionOptions.ROOT_ITSELF;
          }
          library.addRoot(url, rootType, options);
        }
      }
    }
    return library;
  }

  private static <P extends JpsElement> JpsLibrary createLibrary(String name, JpsLibraryPropertiesSerializer<P> loader,
                                                                           final Element propertiesElement) {
    return JpsElementFactory.getInstance().createLibrary(name, loader.getType(), loader.loadProperties(propertiesElement));
  }

  private static JpsOrderRootType getRootType(String rootTypeId) {
    for (JpsLibraryRootTypeSerializer serializer : PREDEFINED_ROOT_TYPES_SERIALIZERS) {
      if (serializer.getTypeId().equals(rootTypeId)) {
        return serializer.getType();
      }
    }
    for (JpsModelSerializerExtension extension : JpsModelSerializerExtension.getExtensions()) {
      for (JpsLibraryRootTypeSerializer serializer : extension.getLibraryRootTypeSerializers()) {
        if (serializer.getTypeId().equals(rootTypeId)) {
          return serializer.getType();
        }
      }
    }
    return JpsOrderRootType.COMPILED;
  }

  private static JpsLibraryPropertiesSerializer<?> getLibraryPropertiesSerializer(@Nullable String typeId) {
    if (typeId != null) {
      for (JpsModelSerializerExtension extension : JpsModelSerializerExtension.getExtensions()) {
        for (JpsLibraryPropertiesSerializer<?> loader : extension.getLibraryPropertiesSerializers()) {
          if (loader.getTypeId().equals(typeId)) {
            return loader;
          }
        }
      }
    }
    return JAVA_LIBRARY_PROPERTIES_SERIALIZER;
  }

  public static JpsElementReference<? extends JpsCompositeElement> createLibraryTableReference(String level) {
    JpsElementFactory elementFactory = JpsElementFactory.getInstance();
    if (level.equals(PROJECT_LEVEL)) {
      return elementFactory.createProjectReference();
    }
    if (level.equals(APPLICATION_LEVEL)) {
      return elementFactory.createGlobalReference();
    }
    for (JpsModelSerializerExtension extension : JpsModelSerializerExtension.getExtensions()) {
      final JpsElementReference<? extends JpsCompositeElement> reference = extension.createLibraryTableReference(level);
      if (reference != null) {
        return reference;
      }
    }
    throw new UnsupportedOperationException();
  }

  public static String getLevelId(JpsElementReference<? extends JpsCompositeElement> reference) {
    if (reference instanceof JpsModuleReference) {
      return MODULE_LEVEL;
    }
    JpsCompositeElement element = reference.resolve();
    if (element instanceof JpsProject) {
      return PROJECT_LEVEL;
    }
    else if (element instanceof JpsGlobal) {
      return APPLICATION_LEVEL;
    }
    for (JpsModelSerializerExtension extension : JpsModelSerializerExtension.getExtensions()) {
      String levelId = extension.getLibraryTableLevelId(reference);
      if (levelId != null) {
        return levelId;
      }
    }
    return null;
  }
}


text:
from copy import deepcopy
import functools
import hashlib
import http.client as httpstatus
import json
from typing import Dict
from typing import List
from typing import Tuple
from unittest import mock
import uuid

from cinder import context
from cinder import db
from cinder import exception
from cinder.tests.unit import test
from cinder.tests.unit import utils as test_utils
from cinder.volume import configuration as conf
from cinder.volume.drivers import lightos

FAKE_LIGHTOS_CLUSTER_NODES: Dict[str, List] = {
    "nodes": [
        {"UUID": "926e6df8-73e1-11ec-a624-000000000001",
         "nvmeEndpoint": "192.168.75.10:4420"},
        {"UUID": "926e6df8-73e1-11ec-a624-000000000002",
         "nvmeEndpoint": "192.168.75.11:4420"},
        {"UUID": "926e6df8-73e1-11ec-a624-000000000003",
         "nvmeEndpoint": "192.168.75.12:4420"}
    ]
}
IPV6_LIST = ['::192:168:75:10', '::192:168:75:11', '::192:168:75:12']
FAKE_LIGHTOS_CLUSTER_NODES_IPV6: Dict[str, List] = {
    "nodes": [
        {"UUID": "926e6df8-73e1-11ec-a624-000000000001",
         "nvmeEndpoint": "[{}]:4420".format(IPV6_LIST[0])},
        {"UUID": "926e6df8-73e1-11ec-a624-000000000002",
         "nvmeEndpoint": "[{}]:4420".format(IPV6_LIST[1])},
        {"UUID": "926e6df8-73e1-11ec-a624-000000000003",
         "nvmeEndpoint": "[{}]:4420".format(IPV6_LIST[2])}
    ]
}

FAKE_LIGHTOS_CLUSTER_INFO: Dict[str, str] = {
    'UUID': "926e6df8-73e1-11ec-a624-07ba3880f6cc",
    'subsystemNQN': "nqn.2014-08.org.nvmexpress:NVMf:uuid:"
    "f4a89ce0-9fc2-4900-bfa3-00ad27995e7b"
}

FAKE_CLIENT_HOSTNQN = "hostnqn1"
VOLUME_BACKEND_NAME = "lightos_backend"
RESERVED_PERCENTAGE = 30
DEVICE_SCAN_ATTEMPTS_DEFAULT = 5
LIGHTOS_API_SERVICE_TIMEOUT = 30
VOLUME_BACKEND_NAME = "lightos_backend"
RESERVED_PERCENTAGE = 30
DEFAULT_COMPRESSION = False

class InitiatorConnectorFactoryMocker:
    @staticmethod
    def factory(protocol, root_helper, driver=None,
                use_multipath=False,
                device_scan_attempts=DEVICE_SCAN_ATTEMPTS_DEFAULT,
                arch=None,
                *args, **kwargs):
        return InitialConnectorMock()

class InitialConnectorMock:
    nqn = FAKE_CLIENT_HOSTNQN
    found_discovery_client = True

    def get_hostnqn(self):
        return self.__class__.nqn

    def find_dsc(self):
        return self.__class__.found_discovery_client

    def get_connector_properties(self, root):
        return dict(nqn=self.__class__.nqn,
                    found_dsc=self.__class__.found_discovery_client)

def get_connector_properties():
    connector = InitialConnectorMock()
    return dict(nqn=connector.get_hostnqn(),
                found_dsc=connector.find_dsc())

def get_vol_etag(volume):
    v = deepcopy(volume)
    v.pop("ETag", None)
    dump = json.dumps(v, sort_keys=True).encode('utf-8')
    return hashlib.md5(dump).hexdigest()

class DBMock(object):

    def __init__(self):
        self.data = {
            "projects": {},
        }

    def get_or_create_project(self, project_name) -> Dict:
        project = self.data["projects"].setdefault(project_name, {})
        return project

    def get_project(self, project_name) -> Dict:
        project = self.data["projects"].get(project_name, None)
        return project if project else None

    def delete_project(self, project_name) -> Dict:
        assert project_name != "default", "can't delete default project"
        project = self.get_project(project_name)
        if not project:
            return None
        self.data["projects"].remove(project)
        return project

    def create_volume(self, volume) -> Tuple[int, Dict]:
        assert volume["project_name"] and volume["name"], "must be provided"
        project = self.get_or_create_project(volume["project_name"])
        volumes = project.setdefault("volumes", [])
        existing_volume = next(iter([vol for vol in volumes
                                     if vol["name"] == volume["name"]]), None)
        if not existing_volume:
            volume["UUID"] = str(uuid.uuid4())
            volumes.append(volume)
            return httpstatus.OK, volume
        return httpstatus.CONFLICT, None

    def get_volume_by_uuid(self, project_name,
                           volume_uuid) -> Tuple[int, Dict]:
        assert project_name and volume_uuid, "must be provided"
        project = self.get_project(project_name)
        if not project:
            return httpstatus.NOT_FOUND, None
        proj_vols = project.get("volumes", None)
        if not proj_vols:
            return httpstatus.NOT_FOUND, None
        vol = next(iter([vol for vol in proj_vols
                         if vol["UUID"] == volume_uuid]), None)
        return (httpstatus.OK, vol) if vol else (httpstatus.NOT_FOUND, None)

    def update_volume_by_uuid(self, project_name,
                              volume_uuid, **kwargs) -> Tuple[int, Dict]:
        error_code, volume = self.get_volume_by_uuid(project_name, volume_uuid)
        if error_code != httpstatus.OK:
            return error_code, None
        etag = kwargs.get("etag", None)
        if etag:
            vol_etag = volume.get("ETag", None)
            if etag != vol_etag:
                return httpstatus.BAD_REQUEST, None
        if kwargs.get("size", None):
            volume["size"] = kwargs["size"]
        if kwargs.get("acl", None):
            volume["acl"] = {'values': kwargs.get('acl')}
        volume["ETag"] = get_vol_etag(volume)
        return httpstatus.OK, volume

    def get_volume_by_name(self, project_name,
                           volume_name) -> Tuple[int, Dict]:
        assert project_name and volume_name, "must be provided"
        project = self.get_project(project_name)
        if not project:
            return httpstatus.NOT_FOUND, None
        proj_vols = project.get("volumes", None)
        if not proj_vols:
            return httpstatus.NOT_FOUND, None
        vol = next(iter([vol for vol in proj_vols
                         if vol["name"] == volume_name]), None)
        return (httpstatus.OK, vol) if vol else (httpstatus.NOT_FOUND, None)

    def delete_volume(self, project_name, volume_uuid) -> Tuple[int, Dict]:
        assert project_name and volume_uuid, "must be provided"
        project = self.get_project(project_name)
        if not project:
            return httpstatus.NOT_FOUND, None
        proj_vols = project.get("volumes", None)
        if not proj_vols:
            return httpstatus.NOT_FOUND, None
        for vol in proj_vols:
            if vol["UUID"] == volume_uuid:
                proj_vols.remove(vol)
        return httpstatus.OK, vol

    def update_volume(self, **kwargs):
        assert("project_name" in kwargs and kwargs["project_name"]), \
            "project_name must be provided"

    def create_snapshot(self, snapshot) -> Tuple[int, Dict]:
        assert snapshot["project_name"] and snapshot["name"], \
            "must be provided"
        project = self.get_or_create_project(snapshot["project_name"])
        snapshots = project.setdefault("snapshots", [])
        existing_snap = next(iter([snap for snap in snapshots
                                   if snap["name"] == snapshot["name"]]), None)
        if not existing_snap:
            snapshot["UUID"] = str(uuid.uuid4())
            snapshots.append(snapshot)
            return httpstatus.OK, snapshot
        return httpstatus.CONFLICT, None

    def delete_snapshot(self, project_name, snapshot_uuid) -> Tuple[int, Dict]:
        assert project_name and snapshot_uuid, "must be provided"
        project = self.get_project(project_name)
        if not project:
            return httpstatus.NOT_FOUND, None
        proj_snaps = project.get("snapshots", None)
        if not proj_snaps:
            return httpstatus.NOT_FOUND, None
        for snap in proj_snaps:
            if snap["UUID"] == snapshot_uuid:
                proj_snaps.remove(snap)
        return httpstatus.OK, snap

    def get_snapshot_by_name(self, project_name,
                             snapshot_name) -> Tuple[int, Dict]:
        assert project_name and snapshot_name, "must be provided"
        project = self.get_project(project_name)
        if not project:
            return httpstatus.NOT_FOUND, None
        proj_snaps = project.get("snapshots", None)
        if not proj_snaps:
            return httpstatus.NOT_FOUND, None
        snap = next(iter([snap for snap in proj_snaps
                          if snap["name"] == snapshot_name]), None)
        return (httpstatus.OK, snap) if snap else (httpstatus.NOT_FOUND, None)

    def get_snapshot_by_uuid(self, project_name,
                             snapshot_uuid) -> Tuple[int, Dict]:
        assert project_name and snapshot_uuid, "must be provided"
        project = self.get_project(project_name)
        if not project:
            return httpstatus.NOT_FOUND, None
        proj_snaps = project.get("snapshots", None)
        if not proj_snaps:
            return httpstatus.NOT_FOUND, None
        snap = next(iter([snap for snap in proj_snaps
                          if snap["UUID"] == snapshot_uuid]), None)
        return (httpstatus.OK, snap) if snap else (httpstatus.NOT_FOUND, None)

class LightOSStorageVolumeDriverTest(test.TestCase):

    def setUp(self):
        """Initialize LightOS Storage Driver."""
        super(LightOSStorageVolumeDriverTest, self).setUp()

        configuration = mock.Mock(conf.Configuration)

        configuration.lightos_api_address = \
            "10.10.10.71,10.10.10.72,10.10.10.73"
        configuration.lightos_api_port = 443
        configuration.lightos_jwt = None
        configuration.lightos_snapshotname_prefix = 'openstack_'
        configuration.lightos_intermediate_snapshot_name_prefix = 'for_clone_'
        configuration.lightos_default_compression_enabled = (
            DEFAULT_COMPRESSION)
        configuration.lightos_default_num_replicas = 3
        configuration.num_volume_device_scan_tries = (
            DEVICE_SCAN_ATTEMPTS_DEFAULT)
        configuration.lightos_api_service_timeout = LIGHTOS_API_SERVICE_TIMEOUT
        configuration.driver_ssl_cert_verify = False
        # for some reason this value is not initialized by the driver parent
        # configs
        configuration.volume_name_template = 'volume-%s'
        configuration.initiator_connector = (
            "cinder.tests.unit.volume.drivers.lightos."
            "test_lightos_storage.InitiatorConnectorFactoryMocker")
        configuration.volume_backend_name = VOLUME_BACKEND_NAME
        configuration.reserved_percentage = RESERVED_PERCENTAGE

        def mocked_safe_get(config, variable_name):
            if hasattr(config, variable_name):
                return config.__getattribute__(variable_name)
            else:
                return None

        configuration.safe_get = functools.partial(mocked_safe_get,
                                                   configuration)
        self.driver = lightos.LightOSVolumeDriver(configuration=configuration)
        self.ctxt = context.get_admin_context()
        self.db: DBMock = DBMock()

        # define a default send_cmd override to return default values.
        def send_cmd_default_mock(cmd, timeout, **kwargs):
            if cmd == "get_nodes":
                return (httpstatus.OK, FAKE_LIGHTOS_CLUSTER_NODES)
            if cmd == "get_node":
                self.assertTrue(kwargs["UUID"])
                for node in FAKE_LIGHTOS_CLUSTER_NODES["nodes"]:
                    if kwargs["UUID"] == node["UUID"]:
                        return (httpstatus.OK, node)
                return (httpstatus.NOT_FOUND, node)
            elif cmd == "get_cluster_info":
                return (httpstatus.OK, FAKE_LIGHTOS_CLUSTER_INFO)
            elif cmd == "create_volume":
                project_name = kwargs["project_name"]
                volume = {
                    "project_name": project_name,
                    "name": kwargs["name"],
                    "size": kwargs["size"],
                    "n_replicas": kwargs["n_replicas"],
                    "compression": kwargs["compression"],
                    "src_snapshot_name": kwargs["src_snapshot_name"],
                    "acl": {'values': kwargs.get('acl')},
                    "state": "Available",
                }
                volume["ETag"] = get_vol_etag(volume)
                code, new_vol = self.db.create_volume(volume)
                return (code, new_vol)
            elif cmd == "delete_volume":
                return self.db.delete_volume(kwargs["project_name"],
                                             kwargs["volume_uuid"])
            elif cmd == "get_volume":
                return self.db.get_volume_by_uuid(kwargs["project_name"],
                                                  kwargs["volume_uuid"])
            elif cmd == "get_volume_by_name":
                return self.db.get_volume_by_name(kwargs["project_name"],
                                                  kwargs["volume_name"])
            elif cmd == "extend_volume":
                size = kwargs.get("size", None)
                return self.db.update_volume_by_uuid(kwargs["project_name"],
                                                     kwargs["volume_uuid"],
                                                     size=size)
            elif cmd == "create_snapshot":
                snapshot = {
                    "project_name": kwargs.get("project_name", None),
                    "name": kwargs.get("name", None),
                    "state": "Available",
                }
                return self.db.create_snapshot(snapshot)
            elif cmd == "delete_snapshot":
                return self.db.delete_snapshot(kwargs["project_name"],
                                               kwargs["snapshot_uuid"])
            elif cmd == "get_snapshot":
                return self.db.get_snapshot_by_uuid(kwargs["project_name"],
                                                    kwargs["snapshot_uuid"])
            elif cmd == "get_snapshot_by_name":
                return self.db.get_snapshot_by_name(kwargs["project_name"],
                                                    kwargs["snapshot_name"])
            elif cmd == "update_volume":
                return self.db.update_volume_by_uuid(**kwargs)

            else:
                raise RuntimeError(
                    f"'{cmd}' is not implemented. kwargs: {kwargs}")

        self.driver.cluster.send_cmd = send_cmd_default_mock

    def test_setup_should_fail_if_lightos_client_cant_auth_cluster(self):
        """Verify lightos_client fail with bad auth."""

        def side_effect(cmd, timeout):
            if cmd == "get_cluster_info":
                return (httpstatus.UNAUTHORIZED, None)
            else:
                raise RuntimeError()

        self.driver.cluster.send_cmd = side_effect
        self.assertRaises(exception.InvalidAuthKey,
                          self.driver.do_setup, None)

    def test_setup_should_succeed(self):
        """Test that lightos_client succeed."""
        self.driver.do_setup(None)

    def test_create_volume_should_succeed(self):
        """Test that lightos_client succeed."""
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)

        self.driver.create_volume(volume)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_create_volume_same_volume_twice_succeed(self):
        """Test succeed to create an exiting volume."""
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)

        self.driver.create_volume(volume)
        self.driver.create_volume(volume)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def _create_volume_in_failed_state(self, vol_state):
        """Verify scenario of created volume in failed state:

        Driver is expected to issue a deletion command and raise exception
        """
        def send_cmd_mock(cmd, **kwargs):
            if cmd == "create_volume":
                project_name = kwargs["project_name"]
                volume = {
                    "project_name": project_name,
                    "name": kwargs["name"],
                    "size": kwargs["size"],
                    "n_replicas": kwargs["n_replicas"],
                    "compression": kwargs["compression"],
                    "src_snapshot_name": kwargs["src_snapshot_name"],
                    "acl": {'values': kwargs.get('acl')},
                    "state": vol_state,
                }
                volume["ETag"] = get_vol_etag(volume)
                code, new_vol = self.db.create_volume(volume)
                return (code, new_vol)
            else:
                return cluster_send_cmd(cmd, **kwargs)

        self.driver.do_setup(None)
        cluster_send_cmd = deepcopy(self.driver.cluster.send_cmd)
        self.driver.cluster.send_cmd = send_cmd_mock
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.driver.create_volume, volume)
        proj = self.db.data["projects"][lightos.LIGHTOS_DEFAULT_PROJECT_NAME]
        actual_volumes = proj["volumes"]
        self.assertEqual(0, len(actual_volumes))
        db.volume_destroy(self.ctxt, volume.id)

    def test_create_volume_in_failed_state(self):
        self._create_volume_in_failed_state("Failed")

    def test_create_volume_in_rollback_state(self):
        self._create_volume_in_failed_state("Rollback")

    def test_create_volume_in_migrating_state_succeed(self):
        """Verify scenario of created volume in migrating state:

        Driver is expected to succeed.
        """
        def send_cmd_mock(cmd, **kwargs):
            if cmd == "create_volume":
                project_name = kwargs["project_name"]
                volume = {
                    "project_name": project_name,
                    "name": kwargs["name"],
                    "size": kwargs["size"],
                    "n_replicas": kwargs["n_replicas"],
                    "compression": kwargs["compression"],
                    "src_snapshot_name": kwargs["src_snapshot_name"],
                    "acl": {'values': kwargs.get('acl')},
                    "state": "Migrating",
                }
                volume["ETag"] = get_vol_etag(volume)
                code, new_vol = self.db.create_volume(volume)
                return (code, new_vol)
            else:
                return cluster_send_cmd(cmd, **kwargs)
        self.driver.do_setup(None)
        cluster_send_cmd = deepcopy(self.driver.cluster.send_cmd)
        self.driver.cluster.send_cmd = send_cmd_mock
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        proj = self.db.data["projects"][lightos.LIGHTOS_DEFAULT_PROJECT_NAME]
        actual_volumes = proj["volumes"]
        self.assertEqual(1, len(actual_volumes))
        db.volume_destroy(self.ctxt, volume.id)

    def test_delete_volume_fail_if_not_created(self):
        """Test that lightos_client fail creating an already exists volume."""
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_extend_volume_should_succeed(self):
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)

        self.driver.create_volume(volume)
        self.driver.extend_volume(volume, 6)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_get_volume_specs_compression_True(self):
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(
            self.ctxt, self,
            extra_specs={'compression': 'True'},
            name='my_vol_typ1')

        vol_type2 = test_utils.create_volume_type(
            self.ctxt, self,
            extra_specs={'compression': '<is> True'},
            name='my_vol_type2')

        vol_type3 = test_utils.create_volume_type(
            self.ctxt, self,
            name='my_vol_type3')

        volume1 = test_utils.create_volume(self.ctxt, size=4,
                                           volume_type_id=vol_type.id)
        volume2 = test_utils.create_volume(self.ctxt, size=4,
                                           volume_type_id=vol_type2.id)
        volume3 = test_utils.create_volume(self.ctxt, size=4,
                                           volume_type_id=vol_type3.id)
        compression, _, _ = self.driver._get_volume_specs(volume1)
        self.assertTrue(compression == "True")
        compression, _, _ = self.driver._get_volume_specs(volume2)
        self.assertTrue(compression == "True")
        compression, _, _ = self.driver._get_volume_specs(volume3)
        self.assertTrue(compression == "False")

        db.volume_destroy(self.ctxt, volume1.id)
        db.volume_destroy(self.ctxt, volume2.id)
        db.volume_destroy(self.ctxt, volume3.id)

    def test_get_volume_specs_compression_False(self):
        self.driver.do_setup(None)
        self.driver.configuration.lightos_default_compression_enabled = True
        vol_type = test_utils.create_volume_type(
            self.ctxt, self,
            extra_specs={'compression': 'False'},
            name='my_vol_typ1')

        vol_type2 = test_utils.create_volume_type(
            self.ctxt, self,
            extra_specs={'compression': '<is> False'},
            name='my_vol_type2')

        vol_type3 = test_utils.create_volume_type(
            self.ctxt, self,
            name='my_vol_type3')
        volume1 = test_utils.create_volume(self.ctxt, size=4,
                                           volume_type_id=vol_type.id)
        volume2 = test_utils.create_volume(self.ctxt, size=4,
                                           volume_type_id=vol_type2.id)
        volume3 = test_utils.create_volume(self.ctxt, size=4,
                                           volume_type_id=vol_type3.id)
        compression, _, _ = self.driver._get_volume_specs(volume1)
        self.assertTrue(compression == "False")
        compression, _, _ = self.driver._get_volume_specs(volume2)
        self.assertTrue(compression == "False")
        compression, _, _ = self.driver._get_volume_specs(volume3)
        self.assertTrue(compression == "True")

        db.volume_destroy(self.ctxt, volume1.id)
        db.volume_destroy(self.ctxt, volume2.id)
        db.volume_destroy(self.ctxt, volume3.id)

    def test_extend_volume_should_fail_if_volume_does_not_exist(self):
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)

        self.assertRaises(exception.VolumeNotFound,
                          self.driver.extend_volume, volume, 6)
        db.volume_destroy(self.ctxt, volume.id)

    def test_create_snapshot(self):
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        snapshot = test_utils.create_snapshot(self.ctxt, volume_id=volume.id)

        self.driver.create_volume(volume)
        self.driver.create_snapshot(snapshot)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_delete_snapshot(self):
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        snapshot = test_utils.create_snapshot(self.ctxt, volume_id=volume.id)

        self.driver.create_volume(volume)
        self.driver.create_snapshot(snapshot)
        self.driver.delete_snapshot(snapshot)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_create_volume_from_snapshot(self):
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        snapshot = test_utils.create_snapshot(self.ctxt, volume_id=volume.id)
        self.driver.create_volume_from_snapshot(volume, snapshot)
        proj = self.db.data["projects"][lightos.LIGHTOS_DEFAULT_PROJECT_NAME]
        actual_volumes = proj["volumes"]
        self.assertEqual(1, len(actual_volumes))
        self.driver.delete_snapshot(snapshot)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)
        db.snapshot_destroy(self.ctxt, snapshot.id)

    def test_initialize_connection(self):
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        connection_props = \
            self.driver.initialize_connection(volume,
                                              get_connector_properties())
        self.assertIn('driver_volume_type', connection_props)
        self.assertEqual('lightos', connection_props['driver_volume_type'])
        self.assertEqual(FAKE_LIGHTOS_CLUSTER_INFO['subsystemNQN'],
                         connection_props['data']['subsysnqn'])
        self.assertEqual(
            self.db.data['projects']['default']['volumes'][0]['UUID'],
            connection_props['data']['uuid'])

        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_initialize_connection_mirgrating_volume(self):
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = True

        def send_cmd_mock(cmd, **kwargs):
            if cmd == "create_volume":
                project_name = kwargs["project_name"]
                volume = {
                    "project_name": project_name,
                    "name": kwargs["name"],
                    "size": kwargs["size"],
                    "n_replicas": kwargs["n_replicas"],
                    "compression": kwargs["compression"],
                    "src_snapshot_name": kwargs["src_snapshot_name"],
                    "acl": {'values': kwargs.get('acl')},
                    "state": "Migrating",
                }
                volume["ETag"] = get_vol_etag(volume)
                code, new_vol = self.db.create_volume(volume)
                return (code, new_vol)
            else:
                return cluster_send_cmd(cmd, **kwargs)
        self.driver.do_setup(None)
        cluster_send_cmd = deepcopy(self.driver.cluster.send_cmd)
        self.driver.cluster.send_cmd = send_cmd_mock
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        connection_props = (
            self.driver.initialize_connection(volume,
                                              get_connector_properties()))
        self.assertIn('driver_volume_type', connection_props)
        self.assertEqual('lightos', connection_props['driver_volume_type'])
        self.assertEqual(FAKE_LIGHTOS_CLUSTER_INFO['subsystemNQN'],
                         connection_props['data']['subsysnqn'])
        self.assertEqual(
            self.db.data['projects']['default']['volumes'][0]['UUID'],
            connection_props['data']['uuid'])

        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_initialize_connection_ipv6(self):
        def side_effect(cmd, timeout, **kwargs):
            if cmd == "get_nodes":
                return (httpstatus.OK, FAKE_LIGHTOS_CLUSTER_NODES_IPV6)
            else:
                return cluster_send_cmd(cmd, timeout, **kwargs)
        cluster_send_cmd = deepcopy(self.driver.cluster.send_cmd)
        self.driver.cluster.send_cmd = side_effect
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        connection_props = (
            self.driver.initialize_connection(volume,
                                              get_connector_properties()))
        self.assertIn('driver_volume_type', connection_props)
        self.assertEqual('lightos', connection_props['driver_volume_type'])
        self.assertEqual(FAKE_LIGHTOS_CLUSTER_INFO['subsystemNQN'],
                         connection_props['data']['subsysnqn'])
        self.assertEqual(
            self.db.data['projects']['default']['volumes'][0]['UUID'],
            connection_props['data']['uuid'])
        for connection in connection_props['data']['lightos_nodes']:
            self.assertIn(connection, IPV6_LIST)

        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_initialize_connection_no_hostnqn_should_fail(self):
        InitialConnectorMock.nqn = ""
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.driver.initialize_connection, volume,
                          get_connector_properties())
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_initialize_connection_no_dsc_should_fail(self):
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = False
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.driver.initialize_connection, volume,
                          get_connector_properties())
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_terminate_connection_with_hostnqn(self):
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        self.driver.terminate_connection(volume, get_connector_properties())
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_terminate_connection_with_empty_hostnqn_should_fail(self):
        InitialConnectorMock.nqn = ""
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.driver.terminate_connection, volume,
                          get_connector_properties())
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_force_terminate_connection_with_empty_hostnqn(self):
        InitialConnectorMock.nqn = ""
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        self.driver.create_volume(volume)
        self.driver.terminate_connection(volume, get_connector_properties(),
                                         force=True)
        self.driver.delete_volume(volume)
        db.volume_destroy(self.ctxt, volume.id)

    def test_check_for_setup_error(self):
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        self.driver.check_for_setup_error()

    def test_check_for_setup_error_no_subsysnqn_should_fail(self):
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        self.driver.cluster.subsystemNQN = ""
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.driver.check_for_setup_error)

    def test_check_for_setup_error_no_hostnqn_should_fail(self):
        InitialConnectorMock.nqn = ""
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.driver.check_for_setup_error)

    def test_check_ip_format(self):
        InitialConnectorMock.nqn = ""
        InitialConnectorMock.found_discovery_client = True
        self.driver.do_setup(None)
        host = "1.1.1.1"
        port = 8009
        endpoint = self.driver.cluster._format_endpoint(host, port)
        self.assertEqual("1.1.1.1:8009", endpoint)
        host = "::1111"
        endpoint = self.driver.cluster._format_endpoint(host, port)
        self.assertEqual("[::1111]:8009", endpoint)

    def test_check_for_setup_error_no_dsc_should_succeed(self):
        InitialConnectorMock.nqn = "hostnqn1"
        InitialConnectorMock.found_discovery_client = False
        self.driver.do_setup(None)
        self.driver.check_for_setup_error()

    def test_create_clone(self):
        self.driver.do_setup(None)

        vol_type = test_utils.create_volume_type(self.ctxt, self,
                                                 name='my_vol_type')
        volume = test_utils.create_volume(self.ctxt, size=4,
                                          volume_type_id=vol_type.id)
        clone = test_utils.create_volume(self.ctxt, size=4,
                                         volume_type_id=vol_type.id)

        self.driver.create_volume(volume)
        self.driver.create_cloned_volume(clone, volume)
        self.driver.delete_volume(volume)
        self.driver.delete_volume(clone)

        db.volume_destroy(self.ctxt, volume.id)
        db.volume_destroy(self.ctxt, clone.id)

    def test_get_volume_stats(self):
        """Test that lightos_client succeed."""
        self.driver.do_setup(None)
        volumes_data = self.driver.get_volume_stats(refresh=False)
        assert len(volumes_data) == 0, "Expected empty config"

        volumes_data = self.driver.get_volume_stats(refresh=True)
        assert volumes_data['vendor_name'] == 'LightOS Storage', \
            "Expected 'LightOS Storage', received %s" % \
            volumes_data['vendor_name']
        assert volumes_data['volume_backend_name'] == VOLUME_BACKEND_NAME, \
            "Expected %s, received %s" % \
            (VOLUME_BACKEND_NAME, volumes_data['volume_backend_name'])
        assert volumes_data['driver_version'] == self.driver.VERSION, \
            "Expected %s, received %s" % \
            (self.driver.VERSION, volumes_data['driver_version'])
        assert volumes_data['storage_protocol'] == "lightos", \
            "Expected 'lightos', received %s" % \
            volumes_data['storage_protocol']
        assert volumes_data['reserved_percentage'] == RESERVED_PERCENTAGE, \
            "Expected %d, received %s" % \
            (RESERVED_PERCENTAGE, volumes_data['reserved_percentage'])
        assert volumes_data['QoS_support'] is False, \
            "Expected False, received %s" % volumes_data['QoS_support']
        assert volumes_data['online_extend_support'] is True, \
            "Expected True, received %s" % \
            volumes_data['online_extend_support']
        assert volumes_data['thin_provisioning_support'] is True, \
            "Expected True, received %s" % \
            volumes_data['thin_provisioning_support']
        assert volumes_data['compression'] == [True, False], \
            "Expected [True, False], received %s" % volumes_data['compression']
        assert volumes_data['multiattach'] is True, \
            "Expected True, received %s" % volumes_data['multiattach']
        assert volumes_data['free_capacity_gb'] == 'infinite', \
            "Expected 'infinite', received %s" % \
            volumes_data['free_capacity_gb']